| threaddb 2.0b A file mapped memory container extension

threaddbSTLWrapper.h
Go to the documentation of this file.
1 /*
2 Copyright (c) 2019 by The ThreadDB Project
3 All Rights Reserved.
4
5 ThreadDB undergoes the BSD License 2.0. You should have received a copy along with this program; if not, write to the ThreadDB Project.
6 To obtain a full unlimited version contact thethreaddbproject(at)gmail.com.
7
8 threaddbSTLWraper.h - standard library container support wrapper classes
9 */
10
11 #pragma once
12
13 #include "threaddbCPP.h"
14 #include <atomic>
15 #include <mutex>
16 #include <vector>
17 #include <deque>
18 #include <memory>
19 #include <stdio.h>
20 #include <string.h>
21
33 namespace tdb
34 {
39 template<class T>
40 struct size_
41 {
42 size_() {}
43 size_(const T&) {}
44 size_t size() const { return sizeof(T); }
45 };
46
51 template<>
52 struct size_<std::string>
53 {
54 size_() : m_size(0) {}
55 size_(const std::string& rS_p) : m_size(rS_p.size()) {}
56 size_t size() const { return m_size; }
57 size_t m_size;
58 };
59
64 template<class T>
65 class tostream
66 {
67 public:
68 tostream(const T& rVal_p) :
69 m_cptr((const char*)&rVal_p)
70 {
71 }
72
73 tostream(const T& rVal_p, size_t /*size_p*/) :
74 m_cptr((const char*)&rVal_p)
75 {
76 }
77
78 const char* get() const
79 {
80 return m_cptr;
81 }
82
83 size_t size() const
84 {
85 return sizeof(T);
86 }
87
88 private:
89 const char* m_cptr;
90 };
91
96 template<class T>
97 class fromstream
98 {
99 public:
100 fromstream()
101 {
102 }
103
104 fromstream(size_<std::string> size_p)
105 {
106 }
107
108 fromstream(database* /*pDb_p*/, ReadInfo* /*pReadInfo_p*/)
109 {
110 }
111
112 std::unique_ptr<T> get(database* pDb_p, ItemInfo* pItemInfo_p, uint64_t Package_p) const
113 {
114 std::unique_ptr<T> pT(new T());
115 pDb_p->Recover(sizeof(T), (char*)pT.get(), *pItemInfo_p, Package_p);
116 return pT;
117 }
118
119 std::unique_ptr<T> get(database* pDb_p, ItemInfo* pItemInfo_p, uint64_t Package_p, size_<T> /*Size_p*/) const
120 {
121 return get(pDb_p, pItemInfo_p, Package_p);
122 }
123
124 std::unique_ptr<T> get(database* pDb_p, ReadInfo* pReadInfo_p) const
125 {
126 std::unique_ptr<T> pT(new T());
127 pDb_p->Recover(sizeof(T), (char*)pT.get(), *pReadInfo_p);
128 return pT;
129 }
130
131 std::unique_ptr<std::string> get() const
132 {
133 return std::unique_ptr<std::string>(new std::string());
134 }
135
136 size_t size() const
137 {
138 return sizeof(T);
139 }
140 };
141
146 template<>
147 class tostream<std::string>
148 {
149 public:
150 tostream(const std::string& rVal_p) :
151 m_size(sizeof(size_t) + rVal_p.size()),
152 m_cptr((char*)malloc(m_size)),
153 m_alloc(true)
154 {
155 *((size_t*)m_cptr) = rVal_p.size();
156 memcpy((void*)(m_cptr + sizeof(size_t)), (void*)rVal_p.c_str(), rVal_p.size());
157 }
158
159 tostream(const std::string& rVal_p, size_<std::string> size_p) :
160 m_size(size_p.size()),
161 m_cptr(rVal_p.c_str()),
162 m_alloc(false)
163 {
164 }
165
166 ~tostream()
167 {
168 if (m_alloc)
169 free((void*)m_cptr);
170 }
171
172 const char* get() const
173 {
174 return m_cptr;
175 }
176
177 size_t size() const
178 {
179 return m_size;
180 }
181
182 private:
183 const char* m_cptr;
184 size_t m_size;
185 bool m_alloc;
186 };
187
192 template<>
193 class fromstream<std::string>
194 {
195 public:
196 fromstream() :
197 m_size()
198 {
199 }
200
201 fromstream(size_<std::string> size_p) :
202 m_size(size_p.size())
203 {
204 }
205
206 fromstream(database* pDb_p, ReadInfo* pReadInfo_p) :
207 m_size()
208 {
209 pDb_p->Recover(sizeof(m_size), (char*)&m_size, *pReadInfo_p);
210 }
211
212 std::unique_ptr<std::string> get(database* pDb_p, ItemInfo* pItemInfo_p, uint64_t Package_p) const
213 {
214 size_t size_ = m_size + sizeof(m_size);
215
216 std::vector<char> cbuf(size_);
217 pDb_p->Recover(size_, (char*)&cbuf[0], *pItemInfo_p, Package_p);
218
219 std::unique_ptr<std::string> pstr(new std::string());
220 (*pstr).insert((*pstr).begin(), &cbuf[sizeof(m_size)], &cbuf[size_]);
221
222 return pstr;
223 }
224
225 std::unique_ptr<std::string> get(database* pDb_p, ItemInfo* pItemInfo_p, uint64_t Package_p, size_<std::string> Size_p) const
226 {
227 std::unique_ptr<std::string> pstr(new std::string());
228 pstr->resize(Size_p.size());
229 pDb_p->Recover(Size_p.size(), (char*)pstr->c_str(), *pItemInfo_p, Package_p);
230 return pstr;
231 }
232
233 std::unique_ptr<std::string> get(database* pDb_p, ReadInfo* pReadInfo_p) const
234 {
235 std::unique_ptr<std::string> pstr(new std::string());
236 pstr->resize(m_size);
237 pDb_p->Recover(m_size, (char*)pstr->c_str(), *pReadInfo_p);
238 return pstr;
239 }
240
241 std::unique_ptr<std::string> get() const
242 {
243 return std::unique_ptr<std::string>(new std::string());
244 }
245
246 size_t size() const
247 {
248 return m_size;
249 }
250
251 private:
252 size_t m_size;
253 };
254
255 template<class T>
256 class ritem;
257
258 template<class T>
259 class sitem;
260
261 typedef std::pair<ItemInfo, std::atomic<uint8_t>> ItemHandle;
262
267 template<class T>
268 class pool
269 {
270 public:
271 pool(database& rDb_p, uint32_t ItemLimit_p = std::numeric_limits<uint32_t>::max()) :
272 m_DeleteLock(),
273 m_DeletedItems(),
274 m_rDb(rDb_p),
275 m_ItemIndex(0),
276 m_ItemLimit(ItemLimit_p),
277 m_ThreadCount(rDb_p.GetThreadCount()),
278 m_Pidx(new std::pair<uint64_t, uint32_t>[m_ThreadCount])
279 {
280 for (size_t idx(0); idx < m_ThreadCount; ++idx)
281 {
282 m_Pidx[idx] = std::make_pair(m_rDb.NewPackage(), 0);
283 }
284 }
285
286 ~pool()
287 {
288 std::lock_guard<std::mutex> deleteLock(m_DeleteLock);
289 do
290 {
291 for (auto iter(m_DeletedItems.begin()); iter != m_DeletedItems.end(); )
292 {
293 if ((*iter).first->first.Unregistered())
294 {
295 delete (*iter).first;
296 iter = m_DeletedItems.erase(iter);
297 continue;
298 }
299 ++iter;
300 }
301
302 if (!m_DeletedItems.empty())
303 {
304 m_rDb.Synchronize(m_DeletedItems.front().second);
305 }
306 } while (m_DeletedItems.empty() == false);
307
308 delete[] m_Pidx;
309 }
310
311 private:
312 friend class ritem<T>;
313 friend class sitem<T>;
314
315 const std::pair<uint64_t, uint32_t>&
316 Store(size_t Size_p, const char* pData_p, ItemInfo* pItemInfo_p = 0)
317 {
318 std::pair<uint64_t, uint32_t>& rPidx(m_Pidx[increment()]);
319 m_rDb.Store(rPidx.first, Size_p, pData_p, pItemInfo_p);
320 return rPidx;
321 }
322
323 ItemHandle* Alloc()
324 {
325 ItemHandle* pItem = 0;
326
327 std::lock_guard<std::mutex> deleteLock(m_DeleteLock);
328
329 auto iter = m_DeletedItems.begin();
330 if (iter != m_DeletedItems.end())
331 {
332 if ((*iter).first->first.m_PackageSize != std::numeric_limits<uint32_t>::max())
333 {
334 pItem = (*iter).first;
335 iter = m_DeletedItems.erase(iter);
336 }
337 }
338
339 return pItem != 0 ? pItem : new ItemHandle();
340 }
341
342 void Delete(ItemHandle* pItemInfo_p, uint64_t PackageID_p)
343 {
344 std::lock_guard<std::mutex> deleteLock(m_DeleteLock);
345 if (pItemInfo_p != 0)
346 {
347 if (pItemInfo_p->first.m_PackageSize != std::numeric_limits<uint32_t>::max())
348 m_DeletedItems.push_front(std::make_pair(pItemInfo_p, PackageID_p));
349 else
350 m_DeletedItems.push_back(std::make_pair(pItemInfo_p, PackageID_p));
351 }
352 }
353
354 database& Db() { return m_rDb; }
355
356 uint32_t increment()
357 {
358 ++m_ItemIndex;
359
360 const uint32_t packageIdx = m_ItemIndex % m_ThreadCount;
361 std::pair<uint64_t, uint32_t>& rPackageIdx(m_Pidx[packageIdx]);
362
363 ++(rPackageIdx.second);
364 if (rPackageIdx.second > m_ItemLimit)
365 {
366 rPackageIdx = std::make_pair(m_rDb.NewPackage(), 1);
367 }
368
369 return packageIdx;
370 }
371
372 std::mutex m_DeleteLock;
373 std::deque< std::pair<ItemHandle*, uint64_t> > m_DeletedItems;
374
375 database& m_rDb;
376 uint32_t m_ItemIndex;
377 uint32_t m_ItemLimit;
378 size_t m_ThreadCount;
379 std::pair<uint64_t, uint32_t>* m_Pidx;
380 };
381
386 template<class T>
387 class ritem
388 {
389 public:
390 typedef T value_type;
391
392 ritem() :
393 m_T(),
394 m_P(),
395 m_Pidx(),
396 m_Size()
397 {
398 m_Pidx = 0;
399 }
400
401 ritem(const T& rVal_p, pool<T>& rPool_p) :
402 m_T(rPool_p.Alloc()),
403 m_P(&rPool_p),
404 m_Pidx(),
405 m_Size(size_<T>(rVal_p))
406 {
407 ++m_T->second;
408 const tostream<T> byteStream(rVal_p, m_Size);
409 m_Pidx = m_P->Store(byteStream.size(), byteStream.get(), &m_T->first).first;
410 }
411
412 ritem(const ritem& rhs_p) :
413 m_T(rhs_p.m_T),
414 m_P(rhs_p.m_P),
415 m_Pidx(rhs_p.m_Pidx),
416 m_Size(rhs_p.m_Size)
417 {
418 ++m_T->second;
419 }
420
421 void operator=(const ritem &rhs_p)
422 {
423 deref();
424
425 m_T = rhs_p.m_T;
426 ++m_T->second;
427 m_P = rhs_p.m_P;
428 m_Pidx = rhs_p.m_Pidx;
429 }
430
431 ~ritem()
432 {
433 deref();
434 }
435
436 void deref()
437 {
438 --m_T->second;
439 if (m_T->second == 0)
440 {
441 m_P->Delete(m_T, m_Pidx);
442 m_T = 0;
443 }
444 }
445
446 std::unique_ptr<T> get() const
447 {
448 return fromstream<T>(m_Size).get(&m_P->Db(), &m_T->first, m_Pidx, m_Size);
449 }
450
451 protected:
452 ItemHandle* m_T;
453
454 private:
455 pool<T>* m_P;
456 uint64_t m_Pidx;
457 size_<T> m_Size;
458 };
459
464 template<class T>
465 class sitem
466 {
467 public:
468 typedef T value_type;
469
470 sitem() :
471 m_P(),
472 m_Pidx()
473 {
474 }
475
476 sitem(const T& rVal_p, pool<T>& rPool_p) :
477 m_P(&rPool_p),
478 m_Pidx()
479 {
480 const tostream<T> byteStream(rVal_p);
481 m_Pidx = m_P->Store(byteStream.size(), byteStream.get());
482 }
483
484 void operator=(const sitem &rhs_p)
485 {
486 m_P = rhs_p.m_P;
487 m_Pidx = rhs_p.m_Pidx;
488 }
489
490 std::unique_ptr<T> get() const
491 {
492 if (!m_P) fromstream<T>().get();
493
494 m_P->Db().Synchronize(m_Pidx.first);
495
496 ReadInfo readInfo = m_P->Db().Open(m_Pidx.first);
497 for (size_t idx(1); idx < m_Pidx.second; ++idx)
498 {
499 fromstream<T> fromStream(&m_P->Db(), &readInfo);
500 const size_t bytesToRead = fromStream.size();
501 if (m_P->Db().Recover(bytesToRead, 0, readInfo) != bytesToRead)
502 throw std::runtime_error("Invalid read operation in item data recovery.");
503 }
504 return fromstream<T>(&m_P->Db(), &readInfo).get(&m_P->Db(), &readInfo);
505 }
506
507 private:
508 pool<T>* m_P;
509 std::pair<uint64_t, uint32_t> m_Pidx;
510 };
511
516 template<class T>
517 class key : public T
518 {
519 public:
520 key(const typename T::value_type& rVal_p, pool<typename T::value_type>& rPool_p) :
521 T(rVal_p, rPool_p)
522 {
523 }
524
525 ~key()
526 {
527 }
528
529 bool operator<(const T& rhs_p) const
530 {
531 return (*this->get()) < (*rhs_p.get());
532 }
533
534 bool operator>(const T& rhs_p) const
535 {
536 return ((*rhs_p.get() < *this->get()));
537 }
538 };
539 }
tdb::ritem::value_type
T value_type
Definition: threaddbSTLWrapper.h:390
tdb::sitem::get
std::unique_ptr< T > get() const
Definition: threaddbSTLWrapper.h:490
tdb::size_< std::string >::size_
size_(const std::string &rS_p)
Definition: threaddbSTLWrapper.h:55
tdb::fromstream< std::string >::size
size_t size() const
Definition: threaddbSTLWrapper.h:246
tdb::fromstream< std::string >::get
std::unique_ptr< std::string > get(database *pDb_p, ItemInfo *pItemInfo_p, uint64_t Package_p) const
Definition: threaddbSTLWrapper.h:212
tdb::ReadInfo
C++ wrapper class to the threadDB_ReadInfo control structure.
Definition: threaddbCPP.h:70
tdb::fromstream::get
std::unique_ptr< T > get(database *pDb_p, ReadInfo *pReadInfo_p) const
Definition: threaddbSTLWrapper.h:124
tdb::fromstream< std::string >::get
std::unique_ptr< std::string > get(database *pDb_p, ItemInfo *pItemInfo_p, uint64_t Package_p, size_< std::string > Size_p) const
Definition: threaddbSTLWrapper.h:225
tdb::size_::size
size_t size() const
Definition: threaddbSTLWrapper.h:44
tdb::fromstream< std::string >::fromstream
fromstream()
Definition: threaddbSTLWrapper.h:196
tdb::ItemInfo
C++ wrapper class to the threadDB_ItemInfo control structure.
Definition: threaddbCPP.h:52
tdb::tostream< std::string >::tostream
tostream(const std::string &rVal_p, size_< std::string > size_p)
Definition: threaddbSTLWrapper.h:159
threaddbCPP.h
Interface C++ wrapper classes.
tdb::fromstream::fromstream
fromstream()
Definition: threaddbSTLWrapper.h:100
tdb::pool::pool
pool(database &rDb_p, uint32_t ItemLimit_p=std::numeric_limits< uint32_t >::max())
Definition: threaddbSTLWrapper.h:271
tdb::key::key
key(const typename T::value_type &rVal_p, pool< typename T::value_type > &rPool_p)
Definition: threaddbSTLWrapper.h:520
tdb::ritem::ritem
ritem(const T &rVal_p, pool< T > &rPool_p)
Definition: threaddbSTLWrapper.h:401
tdb::tostream::size
size_t size() const
Definition: threaddbSTLWrapper.h:83
tdb::ritem::deref
void deref()
Definition: threaddbSTLWrapper.h:436
tdb::key
Wrapper class for storing key data items.
Definition: threaddbSTLWrapper.h:517
tdb::sitem::operator=
void operator=(const sitem &rhs_p)
Definition: threaddbSTLWrapper.h:484
tdb::tostream
Converter of fixed size data elements into a byte stream representation.
Definition: threaddbSTLWrapper.h:65
tdb::tostream< std::string >::tostream
tostream(const std::string &rVal_p)
Definition: threaddbSTLWrapper.h:150
tdb::database::Store
void Store(uint64_t Package_p, size_t Size_p, const char pData_p[], threadDB_ItemInfo *pItemHandle_p=0)
C++ wrapper to store a data item in the selected package see ThreadDB_Store.
Definition: threaddbCPP.h:234
tdb
Definition: threaddbCPP.h:28
tdb::sitem::value_type
T value_type
Definition: threaddbSTLWrapper.h:468
tdb::size_< std::string >::m_size
size_t m_size
Definition: threaddbSTLWrapper.h:57
tdb::size_::size_
size_()
Definition: threaddbSTLWrapper.h:42
tdb::ritem::m_T
ItemHandle * m_T
Definition: threaddbSTLWrapper.h:452
tdb::database::Synchronize
void Synchronize(uint64_t Package_p)
C++ wrapper to synchronize the package buffers prior reading see ThreadDB_Synchronize.
Definition: threaddbCPP.h:243
tdb::sitem
Wrapper class for storing data items in a sequential fashion.
Definition: threaddbSTLWrapper.h:259
tdb::fromstream
Converter of byte stream into fixed data item representation.
Definition: threaddbSTLWrapper.h:97
tdb::tostream::tostream
tostream(const T &rVal_p)
Definition: threaddbSTLWrapper.h:68
tdb::sitem::sitem
sitem()
Definition: threaddbSTLWrapper.h:470
tdb::size_::size_
size_(const T &)
Definition: threaddbSTLWrapper.h:43
tdb::tostream< std::string >::get
const char * get() const
Definition: threaddbSTLWrapper.h:172
tdb::sitem::sitem
sitem(const T &rVal_p, pool< T > &rPool_p)
Definition: threaddbSTLWrapper.h:476
tdb::fromstream::get
std::unique_ptr< std::string > get() const
Definition: threaddbSTLWrapper.h:131
tdb::ritem::get
std::unique_ptr< T > get() const
Definition: threaddbSTLWrapper.h:446
tdb::fromstream< std::string >::get
std::unique_ptr< std::string > get(database *pDb_p, ReadInfo *pReadInfo_p) const
Definition: threaddbSTLWrapper.h:233
tdb::size_< std::string >::size_
size_()
Definition: threaddbSTLWrapper.h:54
tdb::key::operator>
bool operator>(const T &rhs_p) const
Definition: threaddbSTLWrapper.h:534
tdb::size_< std::string >::size
size_t size() const
Definition: threaddbSTLWrapper.h:56
tdb::fromstream::fromstream
fromstream(size_< std::string > size_p)
Definition: threaddbSTLWrapper.h:104
tdb::tostream< std::string >::size
size_t size() const
Definition: threaddbSTLWrapper.h:177
tdb::ritem::~ritem
~ritem()
Definition: threaddbSTLWrapper.h:431
tdb::fromstream< std::string >::fromstream
fromstream(size_< std::string > size_p)
Definition: threaddbSTLWrapper.h:201
tdb::key::~key
~key()
Definition: threaddbSTLWrapper.h:525
tdb::ritem::ritem
ritem()
Definition: threaddbSTLWrapper.h:392
tdb::ritem::ritem
ritem(const ritem &rhs_p)
Definition: threaddbSTLWrapper.h:412
tdb::ItemHandle
std::pair< ItemInfo, std::atomic< uint8_t > > ItemHandle
Definition: threaddbSTLWrapper.h:259
tdb::fromstream::size
size_t size() const
Definition: threaddbSTLWrapper.h:136
tdb::database
C++ wrapper class of the threadDB file mapped memory container extension.
Definition: threaddbCPP.h:97
tdb::pool
Pool datastructure for keeping track and recycling open ItemHandles.
Definition: threaddbSTLWrapper.h:268
tdb::ritem::operator=
void operator=(const ritem &rhs_p)
Definition: threaddbSTLWrapper.h:421
tdb::fromstream::get
std::unique_ptr< T > get(database *pDb_p, ItemInfo *pItemInfo_p, uint64_t Package_p) const
Definition: threaddbSTLWrapper.h:112
tdb::database::Recover
size_t Recover(size_t Size_p, char pData_p[], tdb::ReadInfo &rReadInfo_p)
C++ wrapper for stream reading of data items see ThreadDB_RecoverContinous.
Definition: threaddbCPP.h:289
tdb::fromstream::fromstream
fromstream(database *, ReadInfo *)
Definition: threaddbSTLWrapper.h:108
tdb::fromstream< std::string >::fromstream
fromstream(database *pDb_p, ReadInfo *pReadInfo_p)
Definition: threaddbSTLWrapper.h:206
tdb::database::NewPackage
uint64_t NewPackage()
C++ wrapper to create and register a threadDB data package see ThreadDB_NewPackage.
Definition: threaddbCPP.h:169
tdb::size_< std::string >
Trait class for size of variable length data items (std::string).
Definition: threaddbSTLWrapper.h:52
tdb::tostream< std::string >::~tostream
~tostream()
Definition: threaddbSTLWrapper.h:166
tdb::fromstream::get
std::unique_ptr< T > get(database *pDb_p, ItemInfo *pItemInfo_p, uint64_t Package_p, size_< T >) const
Definition: threaddbSTLWrapper.h:119
tdb::fromstream< std::string >::get
std::unique_ptr< std::string > get() const
Definition: threaddbSTLWrapper.h:241
tdb::tostream::get
const char * get() const
Definition: threaddbSTLWrapper.h:78
tdb::ritem
Wrapper class for storing data items in a random fashion.
Definition: threaddbSTLWrapper.h:256
tdb::pool::~pool
~pool()
Definition: threaddbSTLWrapper.h:286
tdb::key::operator<
bool operator<(const T &rhs_p) const
Definition: threaddbSTLWrapper.h:529
tdb::tostream::tostream
tostream(const T &rVal_p, size_t)
Definition: threaddbSTLWrapper.h:73
tdb::size_
Trait class for size of fixed length data items.
Definition: threaddbSTLWrapper.h:40