ProtoSlots.cpp
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | #include "ProtoSlots.h" | ||
| 2 | |||
| 3 | 22 | proto_slot_t *claimSlot(proto_slot_t *slots, uint8_t count, uint32_t clientId, bool withSchema) { | |
| 4 | 38 | for(uint8_t i = 0; i < count; i++) { | |
| 5 | 37 | uint32_t expected = 0; | |
| 6 | 74 | if(!slots[i].clientId.compare_exchange_strong(expected, clientId)) continue; | |
| 7 | 21 | slots[i].withSchema = withSchema; | |
| 8 | 21 | slots[i].feeds = 0; | |
| 9 | 21 | return &slots[i]; | |
| 10 | } | ||
| 11 | 1 | return nullptr; | |
| 12 | } | ||
| 13 | 20 | void startSync(proto_slot_t &slot) { slot.wants.fetch_or(WANTS_SYNC); } | |
| 14 | 1 | proto_slot_t *slotOf(proto_slot_t *slots, uint8_t count, uint32_t clientId) { | |
| 15 | 6 | for(uint8_t i = 0; i < count; i++) | |
| 16 | 10 | if(slots[i].clientId.load() == clientId) return &slots[i]; | |
| 17 | 1 | return nullptr; | |
| 18 | } | ||
| 19 | 69 | void freeSlot(proto_slot_t &slot) { | |
| 20 | // Cleared before the slot is free, so the next client starts clean. | ||
| 21 | 69 | slot.wants = 0; | |
| 22 | 69 | slot.feeds = 0; | |
| 23 | 69 | slot.clientId = 0; | |
| 24 | 69 | } | |
| 25 | |||
| 26 | 24 | uint32_t takeWants(proto_slot_t &slot, uint32_t seenId) { | |
| 27 | 24 | const uint32_t wants = slot.wants.exchange(0); | |
| 28 | 48 | if(slot.clientId.load() == seenId) return wants; | |
| 29 | // The client left and another took the slot after loop() read it: what was | ||
| 30 | // taken may be the new client's, so it goes back for the next pass. A bit | ||
| 31 | // the old client left behind does no harm, as the new one wants a sync. | ||
| 32 | 1 | if(wants) slot.wants.fetch_or(wants); | |
| 33 | 1 | return 0; | |
| 34 | } | ||
| 35 | 23 | sync_step_t nextSyncStep(proto_slot_t &slot, uint32_t now) { | |
| 36 | 23 | const uint32_t id = slot.clientId.load(); | |
| 37 | 23 | if(id != slot.loopId) { | |
| 38 | // A client left or another took the slot. | ||
| 39 | 14 | slot.loopId = id; | |
| 40 | 14 | slot.cur = sync_cursor_t(); | |
| 41 | 14 | slot.awaitingNext = false; | |
| 42 | } | ||
| 43 | 23 | if(!id) return sync_step_t::none; | |
| 44 | 22 | const uint32_t wants = takeWants(slot, id); | |
| 45 | 22 | if(wants & (WANTS_SYNC | WANTS_SYNC_AGAIN)) { | |
| 46 | // A client that just connected may skip the schema; one that asks again | ||
| 47 | // never gets it. | ||
| 48 | 14 | slot.cur = beginSync((wants & WANTS_SYNC) && slot.withSchema.load()); | |
| 49 | 14 | return sync_step_t::send; | |
| 50 | } | ||
| 51 | 8 | if((wants & WANTS_NEXT) && slot.awaitingNext) return sync_step_t::send; | |
| 52 | 7 | if(slot.awaitingNext && now - slot.sentAt > PROTO_SYNC_TIMEOUT_MS) { | |
| 53 | 1 | slot.awaitingNext = false; | |
| 54 | 1 | slot.cur = sync_cursor_t(); | |
| 55 | 1 | return sync_step_t::timedOut; | |
| 56 | } | ||
| 57 | 6 | return sync_step_t::none; | |
| 58 | } | ||
| 59 | 7 | void syncFrameSent(proto_slot_t &slot, uint32_t now) { | |
| 60 | 7 | slot.awaitingNext = slot.cur.active(); | |
| 61 | 7 | slot.sentAt = now; | |
| 62 | 7 | } | |
| 63 |