| … | … | @@ -73,15 +73,56 @@ class CogChannel |
| 73 | 73 | { |
| 74 | 74 | Client* client; |
| 75 | 75 | void (Client::*callback)(const std::string&, const Data&); |
| 76 | size_t sequence; |
| 76 | 77 | bool noreply; // If true, skip do_recv() |
| 77 | 78 | |
| 78 | 79 | std::string str_to_send; |
| 79 | 80 | Data data; |
| 80 | 81 | |
| 82 | // Sequence counter for non-idempotent messages |
| 83 | static std::atomic<size_t> _sequence_counter; |
| 84 | |
| 85 | // Default constructor required by concurrent_set |
| 86 | Msg() : client(nullptr), callback(nullptr), sequence(0), noreply(true) {} |
| 87 | |
| 88 | // The mesage buffer is a de-duplicating buffer: identical |
| 89 | // messages are added only once. This makes sense for almost |
| 90 | // all messages: we do not need to store an Atom twice; the |
| 91 | // store operation is idempotent. There are two exceptions. |
| 92 | // One is the cog-value-increment message: these are never |
| 93 | // idempotent; we use a sequence number to make sure each is |
| 94 | // unique. The other case is the barrier. The exact same |
| 95 | // barrier message must be sent to each socket. The barrier |
| 96 | // code will make sure that the queues are all empty before |
| 97 | // enqueuing the barrier, so ordering does not matter, but |
| 98 | // having exactly NTHREADS copies does. It must not be |
| 99 | // deduplicated So that is what this ctor does. |
| 100 | Msg(Client* c, void (Client::*cb)(const std::string&, const Data&), |
| 101 | bool nr, const std::string& str, const Data& d) |
| 102 | : client(c), callback(cb), noreply(nr), |
| 103 | str_to_send(str), data(d) |
| 104 | { |
| 105 | if (str.compare(0, 19, "(cog-update-value!") == 0 or |
| 106 | str.compare(0, 13, "(cog-barrier)") == 0) |
| 107 | { |
| 108 | sequence = ++_sequence_counter; |
| 109 | } |
| 110 | else |
| 111 | { |
| 112 | sequence = 0; |
| 113 | } |
| 114 | } |
| 115 | |
| 81 | 116 | // Need operator<() in order to queue up the messages. |
| 82 | | // XXX Maybe should be `(this->data < other.data)` ?? |
| 83 | | int operator<(const Msg& other) const |
| 84 | | { return this < &other; } |
| 117 | // If both sequence numbers are zero, compare by string content |
| 118 | // (idempotent messages deduplicate). Otherwise compare by |
| 119 | // sequence number (non-idempotent messages stay unique). |
| 120 | bool operator<(const Msg& other) const |
| 121 | { |
| 122 | if (sequence == 0 and other.sequence == 0) |
| 123 | return str_to_send < other.str_to_send; |
| 124 | return sequence < other.sequence; |
| 125 | } |
| 85 | 126 | }; |
| 86 | 127 | |
| 87 | 128 | async_buffer<CogChannel, Msg> _msg_buffer; |