blob: 7292d8be53d9b9f1bd456e5049c4f0e516e449e3 (
plain) (
blame)
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
#pragma once
#include "ring_buffer_with_spin_lock.h"
#include <util/generic/array_ref.h>
#include <util/generic/vector.h>
#include <util/system/condvar.h>
#include <util/system/event.h>
#include <util/system/mutex.h>
#include <util/system/thread.h>
#include <util/thread/lfqueue.h>
namespace NActor {
namespace NPrivate {
struct TExecutorHistory {
struct THistoryRecord {
ui32 MaxQueueSize;
};
TVector<THistoryRecord> HistoryRecords;
ui64 LastHistoryRecordSecond;
ui64 FirstHistoryRecordSecond() const {
return LastHistoryRecordSecond - HistoryRecords.size() + 1;
}
};
struct TExecutorStatus {
size_t WorkQueueSize = 0;
TExecutorHistory History;
TString Status;
};
}
class IWorkItem {
public:
virtual ~IWorkItem() {
}
virtual void DoWork(/* must release this */) = 0;
};
struct TExecutorWorker;
class TExecutor: public TAtomicRefCount<TExecutor> {
friend struct TExecutorWorker;
public:
struct TConfig {
size_t WorkerCount;
const char* Name;
TConfig()
: WorkerCount(1)
, Name()
{
}
};
private:
struct TImpl;
THolder<TImpl> Impl;
const TConfig Config;
TAtomic ExitWorkers;
TVector<TAutoPtr<TExecutorWorker>> WorkerThreads;
TRingBufferWithSpinLock<IWorkItem*> WorkItems;
TMutex WorkMutex;
TCondVar WorkAvailable;
public:
explicit TExecutor(size_t workerCount);
TExecutor(const TConfig& config);
~TExecutor();
void Stop();
void EnqueueWork(TArrayRef<IWorkItem* const> w);
size_t GetWorkQueueSize() const;
TString GetStatus() const;
TString GetStatusSingleLine() const;
NPrivate::TExecutorStatus GetStatusRecordInternal() const;
bool IsInExecutorThread() const;
private:
void Init();
TAutoPtr<IWorkItem> DequeueWork();
void ProcessWorkQueueHere();
inline void RunWorkItem(TAutoPtr<IWorkItem>);
void RunWorker();
ui32 GetMaxQueueSizeAndClear() const;
};
using TExecutorPtr = TIntrusivePtr<TExecutor>;
}
|
