1
epoll_event events[LOWER_LOAD_EVENT_SIZE];
2
while (_event_group.wait_and_handle(events, LOWER_LOAD_EVENT_SIZE, -1)) {
3
    // check pending write queue, and block write all
4
    for (auto& entry : _pending_data) {
5
        // write all requests to provider agent
6
        // ...
7
    }
8
}

1
template <class K, class V, class C = tbb::tbb_hash_compare<K>,
2
          class A = tbb::cache_aligned_allocator<std::pair<const K, V>>>
3
class ConcurrentHashMapProxy : public tbb::concurrent_hash_map<K, V, C, A> {
4
public:
5
    ConcurrentHashMapProxy(size_t n) : tbb::concurrent_hash_map<K, V, C, A>(n) {}
6
    ~ConcurrentHashMapProxy() {
7
        this->tbb::concurrent_hash_map<K, V, C, A>::~concurrent_hash_map<K, V, C, A>();
8
    }
9
    typename tbb::concurrent_hash_map<K, V, C, A>::const_pointer fast_find(const K& key) const {
10
        return this->internal_fast_find(key);
11
    }
12
};
13

14
MessageQueue* QueueStore::find_or_create_queue(const String& queue_name) {
15
    auto q_ptr = queues_.fast_find(queue_name);
16
    if (q_ptr) return q_ptr->second;
17

18
    decltype(queues_)::const_accessor ac;
19
    queues_.find(ac, queue_name);
20
    if (ac.empty()) {
21
        uint32_t queue_id = next_queue_id_.next();
22
        auto queue_ptr = new MessageQueue(queue_id, this);
23
        queues_.insert(ac, std::make_pair(queue_name, queue_ptr));
24
        DLOG("Created a new queue, id: %d, name: %s", q->get_queue_id(),
25
             q->get_queue_name().c_str());
26
    }
27
    return ac->second;
28
}
29

30
MessageQueue* QueueStore::find_queue(const String& queue_name) const {
31
    auto q_ptr = queues_.fast_find(queue_name);
32
    return q_ptr ? q_ptr->second : nullptr;
33

34
    decltype(queues_)::const_accessor ac;
35
    queues_.find(ac, queue_name);
36
    return ac.empty() ? nullptr : ac->second;
37
}
38

39
void QueueStore::put(const String& queue_name, const MemBlock& message) {
40
    if (!message.ptr) return;
41
    auto q_ptr = find_or_create_queue(queue_name);
42
    q_ptr->put(message);
43
}
44

45
Vector<MemBlock> QueueStore::get(const String& queue_name, long offset, long size) {
46
    if (!flushed) flush_all_before_read();
47

48
    auto q_ptr = find_queue(queue_name);
49
    if (q_ptr) return q_ptr->get(offset, size);
50
    // return empty list when queue is not found
51
    return Vector<MemBlock>();
52
}

1
// File page header
2
struct __attribute__((__packed__)) FilePageHeader {
3
    uint64_t offset = NEGATIVE_OFFSET;
4
};
5

6
#define FILE_PAGE_HEADER_SIZE sizeof(FilePageHeader)
7
#define FILE_PAGE_AVAILABLE_SIZE (FILE_PAGE_SIZE - FILE_PAGE_HEADER_SIZE)
8

9
// File page
10
struct __attribute__((__packed__)) FilePage {
11
    FilePageHeader header;
12
    char content[FILE_PAGE_AVAILABLE_SIZE];
13
};

1
// buffer is full, switch to back
2
{
3
    // spin wait until back buf is not in scheduled status
4
    while (back_buf_status->load() == BACK_BUF_FLUSH_SCHEDULED)
5
        ;
6
    // try swap active and back buffer
7
    std::unique_lock<std::mutex> lock(*back_buf_mutex);
8
    std::swap(active_buf, back_buf);
9
    assert(back_buf_status->load() == BACK_BUF_FREE);
10
    uint32_t exp = BACK_BUF_FREE;
11
    back_buf_status->compare_exchange_strong(exp, BACK_BUF_FLUSH_SCHEDULED);
12
}

1
// release free memory back to os
2
MallocExtension::instance()->ReleaseFreeMemory();

1
// first page will hold at most (queue_id / DATA_FILE_SPLITS) % 64 + 1 messages, this make write
2
// more average. This leads to 64 timepoints of first flush. I call it flush fast.
3
bool flush_fast =
4
    paged_message_indices_.size() == 1 &&
5
    paged_message_indices_.back().msg_size >= ((queue_id_ / DATA_FILE_SPLITS) & 0x3f) + 1;