tair-kvcache/kv_cache_manager/event/log_event_publisher.cc at 879e684b608dcebb595607ef6b06db6dd8ce2a16 · alibaba/tair-kvcache · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
#include "kv_cache_manager/event/log_event_publisher.h"

#include <chrono>
#include <filesystem>
#include <iomanip>
#include <iostream>
#include <sstream>

#include "kv_cache_manager/common/jsonizable.h"
#include "kv_cache_manager/common/logger.h"
namespace kv_cache_manager {

LogEventPublisher::LogEventPublisher() : LogEventPublisher(LogEventPublisherConfig{}) {}

LogEventPublisher::LogEventPublisher(const LogEventPublisherConfig &config) : config_(config) {}

LogEventPublisher::~LogEventPublisher() {
    if (running_) {
        Stop();
    }
}
bool LogEventPublisher::Init(const std::string & /*config*/) {
    InitBasicQueue(config_.queue_size());

    running_ = true;

    worker_ = std::thread(&LogEventPublisher::WorkerThread, this);

    return true;
}

bool LogEventPublisher::Publish(const std::shared_ptr<BaseEvent> &event) {
    if (!event) {
        return false;
    }

    if (!running_) {
        return false;
    }
    if (!BasicEnqueue(event)) {
        KVCM_LOG_WARN("Event queue full, dropping event");
        return false;
    }
    KVCM_LOG_DEBUG("Event enqueued successfully");
    return true;
}

bool LogEventPublisher::Stop() {
    if (!running_) {
        return true;
    }
    running_ = false;
    ClearBasicQueue();
    if (basic_queue_) {
        basic_queue_->queue_cv.notify_all();
    }

    if (worker_.joinable()) {
        worker_.join();
    }

    return true;
}

void LogEventPublisher::WorkerThread() {
    while (running_) {
        BasicWait();

        std::shared_ptr<BaseEvent> process_event;
        while (BasicDequeue(process_event)) {
            WriteEventToFile(process_event);
        }
    }
}

void LogEventPublisher::WriteEventToFile(const std::shared_ptr<BaseEvent> &event) {
    if (!event) {
        return;
    }

    KVCM_PUBLISHER_LOG(FormatEvent(event));
}

std::string LogEventPublisher::FormatEvent(const std::shared_ptr<BaseEvent> &event) const {
    if (!event) {
        return "{}";
    }
    return event->ToJsonString();
}

} // namespace kv_cache_manager