Skip to content

Commit e094d11

Browse files
malytomasmalytomas
committed
ring buffer, sliding buffer
1 parent fe27862 commit e094d11

91 files changed

Lines changed: 563 additions & 250 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎sources/include/cage-core/blockContainer.h‎

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@ namespace cage
1212
{
1313
BlockContainer() = default;
1414

15-
CAGE_FORCE_INLINE explicit BlockContainer(uint32 maxBlockSize) : blockSize(maxBlockSize) {}
15+
explicit BlockContainer(uint32 maxBlockSize) : blockSize(maxBlockSize) {}
1616

1717
struct Iterator
1818
{
@@ -48,35 +48,35 @@ namespace cage
4848
CAGE_FORCE_INLINE PointerRange<T> operator->() const { return const_cast<BlockContainer *>(collection)->blocks.at(index); }
4949
CAGE_FORCE_INLINE PointerRange<T> operator*() const { return const_cast<BlockContainer *>(collection)->blocks.at(index); }
5050

51-
private:
51+
protected:
5252
const BlockContainer *collection = nullptr;
5353
uint32 index = m;
5454
};
5555

5656
CAGE_FORCE_INLINE Iterator begin() const { return Iterator(this, 0); }
5757
CAGE_FORCE_INLINE Iterator end() const { return Iterator(this, numeric_cast<uint32>(blocks.size())); }
58-
CAGE_FORCE_INLINE bool empty() const { return size_ == 0; }
59-
CAGE_FORCE_INLINE uint32 size() const { return size_; }
58+
CAGE_FORCE_INLINE bool empty() const noexcept { return size_ == 0; }
59+
CAGE_FORCE_INLINE uint32 size() const noexcept { return size_; }
6060

61-
CAGE_FORCE_INLINE void push_back(const T &val)
61+
void push_back(const T &val)
6262
{
6363
inserting().push_back(val);
6464
size_++;
6565
}
6666

67-
CAGE_FORCE_INLINE void push_back(T &&val)
67+
void push_back(T &&val)
6868
{
6969
inserting().push_back(std::move(val));
7070
size_++;
7171
}
7272

73-
CAGE_FORCE_INLINE void clear()
73+
void clear()
7474
{
7575
blocks.clear();
7676
size_ = 0;
7777
}
7878

79-
private:
79+
protected:
8080
std::vector<std::vector<T>> blocks;
8181
uint32 size_ = 0;
8282
uint32 blockSize = 200;

‎sources/include/cage-core/concurrentQueue.h‎

Lines changed: 14 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,6 @@
11
#ifndef guard_concurrentQueue_h_F17509C840DB4228AF89C97FCD8EC1E5
22
#define guard_concurrentQueue_h_F17509C840DB4228AF89C97FCD8EC1E5
33

4-
#include <vector>
5-
64
#include <cage-core/concurrent.h>
75

86
namespace cage
@@ -12,7 +10,7 @@ namespace cage
1210
using Exception::Exception;
1311
};
1412

15-
template<class T>
13+
template<class T, template<class...> class Container>
1614
class ConcurrentQueue : private Immovable
1715
{
1816
public:
@@ -99,7 +97,7 @@ namespace cage
9997
else
10098
{
10199
value = std::move(items.front());
102-
items.erase(items.begin());
100+
internalPopFront();
103101
writer->signal();
104102
return;
105103
}
@@ -114,7 +112,7 @@ namespace cage
114112
if (!items.empty())
115113
{
116114
value = std::move(items.front());
117-
items.erase(items.begin());
115+
internalPopFront();
118116
writer->signal();
119117
return true;
120118
}
@@ -130,21 +128,29 @@ namespace cage
130128
reader->broadcast();
131129
}
132130

133-
bool stopped() const
131+
CAGE_FORCE_INLINE bool stopped() const
134132
{
135133
ScopeLock sl(mut); // mandate memory barriers
136134
return stop;
137135
}
138136

139-
uint32 estimatedSize() const { return numeric_cast<uint32>(items.size()); }
137+
CAGE_FORCE_INLINE uint32 estimatedSize() const noexcept { return numeric_cast<uint32>(items.size()); }
140138

141139
protected:
142140
Holder<Mutex> mut = newMutex();
143141
Holder<ConditionalVariable> writer = newConditionalVariable();
144142
Holder<ConditionalVariable> reader = newConditionalVariable();
145-
std::vector<T> items;
143+
Container<T> items;
146144
uint32 maxItems = m;
147145
bool stop = false;
146+
147+
CAGE_FORCE_INLINE void internalPopFront()
148+
{
149+
if constexpr (requires(Container<T> c) { c.pop_front(); })
150+
items.pop_front();
151+
else
152+
items.erase(items.begin());
153+
}
148154
};
149155
}
150156

‎sources/include/cage-core/ringBuffer.h‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
#ifndef guard_ringBuffer_h_esr54uguf4olkjhgserdtz
2+
#define guard_ringBuffer_h_esr54uguf4olkjhgserdtz
3+
4+
#include <vector>
5+
6+
#include <cage-core/core.h>
7+
8+
namespace cage
9+
{
10+
template<class T>
11+
class RingBuffer : private Noncopyable
12+
{
13+
public:
14+
CAGE_FORCE_INLINE bool empty() const noexcept { return size_ == 0; }
15+
16+
CAGE_FORCE_INLINE uintPtr size() const noexcept { return size_; }
17+
18+
CAGE_FORCE_INLINE const T &front() const
19+
{
20+
CAGE_ASSERT(!empty());
21+
return items[offset_];
22+
}
23+
24+
CAGE_FORCE_INLINE T &front()
25+
{
26+
CAGE_ASSERT(!empty());
27+
return items[offset_];
28+
}
29+
30+
void push_back(const T &val) { push_back(T(val)); }
31+
32+
void push_back(T &&val)
33+
{
34+
if (size_ == items.size())
35+
{ // must reallocate
36+
std::vector<T> repl;
37+
repl.resize(size_ * 3 / 2 + 10);
38+
repl.resize(repl.capacity()); // make sure to use all of the allocated capacity
39+
const uintPtr over = size_ - offset_;
40+
for (uintPtr i = 0; i < over; i++)
41+
repl[i] = std::move(items[offset_ + i]);
42+
for (uintPtr i = 0; i < offset_; i++)
43+
repl[over + i] = std::move(items[i]);
44+
std::swap(items, repl);
45+
offset_ = 0;
46+
}
47+
48+
items[(offset_ + size_) % items.size()] = std::move(val);
49+
size_++;
50+
}
51+
52+
void pop_front()
53+
{
54+
CAGE_ASSERT(!empty());
55+
items[offset_] = {};
56+
size_--;
57+
if (size_ == 0)
58+
offset_ = 0;
59+
else
60+
offset_ = (offset_ + 1) % items.size();
61+
}
62+
63+
void clear()
64+
{
65+
offset_ = size_ = 0;
66+
items.clear();
67+
}
68+
69+
protected:
70+
std::vector<T> items;
71+
uintPtr offset_ = 0;
72+
uintPtr size_ = 0;
73+
};
74+
}
75+
76+
#endif // guard_ringBuffer_h_esr54uguf4olkjhgserdtz

‎sources/include/cage-core/slidingBuffer.h‎

Lines changed: 119 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,119 @@
1+
#ifndef guard_slidingBuffer_h_srhphd54ujz5u4i6dftrz5d
2+
#define guard_slidingBuffer_h_srhphd54ujz5u4i6dftrz5d
3+
4+
#include <vector>
5+
6+
#include <cage-core/core.h>
7+
8+
namespace cage
9+
{
10+
template<class T>
11+
class SlidingBuffer : private Noncopyable
12+
{
13+
public:
14+
CAGE_FORCE_INLINE const T *begin() const { return items.data() + offset_; }
15+
16+
CAGE_FORCE_INLINE T *begin() { return items.data() + offset_; }
17+
18+
CAGE_FORCE_INLINE const T *end() const { return items.data() + offset_ + size_; }
19+
20+
CAGE_FORCE_INLINE T *end() { return items.data() + offset_ + size_; }
21+
22+
CAGE_FORCE_INLINE bool empty() const noexcept { return size_ == 0; }
23+
24+
CAGE_FORCE_INLINE uintPtr size() const noexcept { return size_; }
25+
26+
CAGE_FORCE_INLINE const T &front() const
27+
{
28+
CAGE_ASSERT(!empty());
29+
return items[offset_];
30+
}
31+
32+
CAGE_FORCE_INLINE T &front()
33+
{
34+
CAGE_ASSERT(!empty());
35+
return items[offset_];
36+
}
37+
38+
void push_back(const T &val) { push_back(T(val)); }
39+
40+
void push_back(T &&val)
41+
{
42+
if (offset_ + size_ < items.capacity())
43+
{ // we have enough space for directly inserting at the end
44+
if (offset_ + size_ >= items.size())
45+
items.push_back(std::move(val));
46+
else
47+
items[offset_ + size_] = std::move(val);
48+
size_++;
49+
return;
50+
}
51+
CAGE_ASSERT(offset_ + size_ == items.size());
52+
53+
if (offset_ * 2 > items.size() + 10)
54+
{ // we prefer to move existing items to front
55+
for (uintPtr i = 0; i < size_; i++)
56+
items[i] = std::move(items[offset_ + i]);
57+
offset_ = 0;
58+
items[size_] = std::move(val);
59+
size_++;
60+
return;
61+
}
62+
63+
// we must reallocate
64+
items.push_back(std::move(val));
65+
size_++;
66+
}
67+
68+
void pop_front()
69+
{
70+
CAGE_ASSERT(!empty());
71+
items[offset_] = {};
72+
size_--;
73+
if (size_ == 0)
74+
offset_ = 0;
75+
else
76+
offset_++;
77+
}
78+
79+
void erase(const T *it)
80+
{
81+
CAGE_ASSERT(!empty());
82+
CAGE_ASSERT(it >= items.data() && it < items.data() + items.size());
83+
const uintPtr index = it - items.data();
84+
CAGE_ASSERT(index >= offset_ && index < offset_ + size_);
85+
items[index] = {};
86+
87+
const uintPtr a = index - offset_;
88+
const uintPtr b = offset_ + size_ - index - 1;
89+
CAGE_ASSERT(a + b + 1 == size_);
90+
if (b <= a)
91+
{ // shift following items left
92+
for (uintPtr i = 0; i < b; i++)
93+
items[index + i] = std::move(items[index + i + 1]);
94+
}
95+
else
96+
{ // shift preceding items right
97+
for (uintPtr i = a; i; i--)
98+
items[offset_ + i] = std::move(items[offset_ + i - 1]);
99+
offset_++;
100+
}
101+
size_--;
102+
if (size_ == 0)
103+
offset_ = 0;
104+
}
105+
106+
void clear()
107+
{
108+
offset_ = size_ = 0;
109+
items.clear();
110+
}
111+
112+
protected:
113+
std::vector<T> items;
114+
uintPtr offset_ = 0;
115+
uintPtr size_ = 0;
116+
};
117+
}
118+
119+
#endif // guard_slidingBuffer_h_srhphd54ujz5u4i6dftrz5d

‎sources/libcore/assets/assetsManager.cpp‎

Lines changed: 5 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
#include <cage-core/networkTcp.h>
2323
#include <cage-core/pointerRangeHolder.h>
2424
#include <cage-core/profiling.h>
25+
#include <cage-core/ringBuffer.h>
2526
#include <cage-core/scopeGuard.h>
2627
#include <cage-core/serialization.h>
2728
#include <cage-core/stdHash.h>
@@ -172,9 +173,9 @@ namespace cage
172173
ankerl::unordered_dense::map<uint32, Collection> privateIndex;
173174
ankerl::unordered_dense::map<uint32, Holder<Asset>> publicIndex;
174175
ankerl::unordered_dense::map<uint32, std::vector<Holder<Waiting>>> waitingIndex;
175-
std::vector<Holder<ConcurrentQueue<Work>>> customProcessingQueues;
176-
ConcurrentQueue<Holder<AsyncTask>> tasksCleanupQueue;
177-
ConcurrentQueue<Work> fetchQueue;
176+
std::vector<Holder<ConcurrentQueue<Work, RingBuffer>>> customProcessingQueues;
177+
ConcurrentQueue<Holder<AsyncTask>, RingBuffer> tasksCleanupQueue;
178+
ConcurrentQueue<Work, RingBuffer> fetchQueue;
178179
std::vector<Holder<Thread>> fetchThreads;
179180
Holder<void> listener;
180181

@@ -184,7 +185,7 @@ namespace cage
184185
schemes.resize(config.schemesMaxCount);
185186
customProcessingQueues.resize(config.customProcessingThreads);
186187
for (auto &it : customProcessingQueues)
187-
it = systemMemory().createHolder<ConcurrentQueue<Work>>();
188+
it = systemMemory().createHolder<ConcurrentQueue<Work, RingBuffer>>();
188189
fetchThreads.reserve(config.diskLoadingThreads);
189190
for (uint32 i = 0; i < config.diskLoadingThreads; i++)
190191
fetchThreads.push_back(newThread(Delegate<void()>().bind<AssetsManagerImpl, &AssetsManagerImpl::diskLoadingEntry>(this), Stringizer() + "asset disk loading " + i));

‎sources/libcore/concurrent/tasks.cpp‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77
#include <cage-core/debug.h>
88
#include <cage-core/profiling.h>
99
#include <cage-core/scopeGuard.h>
10+
#include <cage-core/slidingBuffer.h>
1011
#include <cage-core/tasks.h>
1112

1213
namespace cage
@@ -116,7 +117,7 @@ namespace cage
116117
}
117118
};
118119

119-
struct TasksQueue : public ConcurrentQueue<Holder<TaskImpl>>
120+
struct TasksQueue : public ConcurrentQueue<Holder<TaskImpl>, SlidingBuffer>
120121
{
121122
bool tryPopFilter(Holder<TaskImpl> &value)
122123
{

‎sources/libcore/errors/crashHandlerSignals.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -209,7 +209,7 @@ namespace cage
209209
case SIGXFSZ:
210210
return "SIGXFSZ";
211211
default:
212-
return "UNKNOWN";
212+
return "unknown";
213213
}
214214
}
215215

0 commit comments

Comments
 (0)