mailbox.cpp 23 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590
  1. #include "mailbox.h"
  2. #include "actorsystem.h"
  3. #include "actor.h"
  4. #include <library/cpp/actors/util/datetime.h>
  5. #include <util/system/sanitizers.h>
  6. namespace NActors {
  7. TMailboxTable::TMailboxTable()
  8. : LastAllocatedLine(0)
  9. , AllocatedMailboxCount(0)
  10. , CachedSimpleMailboxes(0)
  11. , CachedRevolvingMailboxes(0)
  12. , CachedHTSwapMailboxes(0)
  13. , CachedReadAsFilledMailboxes(0)
  14. , CachedTinyReadAsFilledMailboxes(0)
  15. {
  16. memset((void*)Lines, 0, sizeof(Lines));
  17. }
  18. bool IsGoodForCleanup(const TMailboxHeader* header) {
  19. switch (AtomicLoad(&header->ExecutionState)) {
  20. case TMailboxHeader::TExecutionState::Inactive:
  21. case TMailboxHeader::TExecutionState::Scheduled:
  22. return true;
  23. case TMailboxHeader::TExecutionState::Leaving:
  24. case TMailboxHeader::TExecutionState::Executing:
  25. case TMailboxHeader::TExecutionState::LeavingMarked:
  26. return false;
  27. case TMailboxHeader::TExecutionState::Free:
  28. case TMailboxHeader::TExecutionState::FreeScheduled:
  29. return true;
  30. case TMailboxHeader::TExecutionState::FreeLeaving:
  31. case TMailboxHeader::TExecutionState::FreeExecuting:
  32. case TMailboxHeader::TExecutionState::FreeLeavingMarked:
  33. return false;
  34. default:
  35. Y_FAIL();
  36. }
  37. }
  38. template <typename TMailbox>
  39. void DestructMailboxLine(ui8* begin, ui8* end) {
  40. const ui32 sx = TMailbox::AlignedSize();
  41. for (ui8* x = begin; x + sx <= end; x += sx) {
  42. TMailbox* mailbox = reinterpret_cast<TMailbox*>(x);
  43. Y_ABORT_UNLESS(IsGoodForCleanup(mailbox));
  44. mailbox->ExecutionState = Max<ui32>();
  45. mailbox->~TMailbox();
  46. }
  47. }
  48. template <typename TMailbox>
  49. bool CleanupMailboxLine(ui8* begin, ui8* end) {
  50. const ui32 sx = TMailbox::AlignedSize();
  51. bool done = true;
  52. for (ui8* x = begin; x + sx <= end; x += sx) {
  53. TMailbox* mailbox = reinterpret_cast<TMailbox*>(x);
  54. Y_ABORT_UNLESS(IsGoodForCleanup(mailbox));
  55. done &= mailbox->CleanupActors() && mailbox->CleanupEvents();
  56. }
  57. return done;
  58. }
  59. TMailboxTable::~TMailboxTable() {
  60. // on cleanup we must traverse everything and free stuff
  61. for (ui32 i = 0; i < LastAllocatedLine; ++i) {
  62. if (TMailboxLineHeader* lineHeader = Lines[i]) {
  63. switch (lineHeader->MailboxType) {
  64. case TMailboxType::Simple:
  65. DestructMailboxLine<TSimpleMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  66. break;
  67. case TMailboxType::Revolving:
  68. DestructMailboxLine<TRevolvingMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  69. break;
  70. case TMailboxType::HTSwap:
  71. DestructMailboxLine<THTSwapMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  72. break;
  73. case TMailboxType::ReadAsFilled:
  74. DestructMailboxLine<TReadAsFilledMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  75. break;
  76. case TMailboxType::TinyReadAsFilled:
  77. DestructMailboxLine<TTinyReadAsFilledMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  78. break;
  79. default:
  80. Y_FAIL();
  81. }
  82. lineHeader->~TMailboxLineHeader();
  83. free(lineHeader);
  84. Lines[i] = nullptr;
  85. }
  86. }
  87. while (MailboxCacheSimple.Pop(0))
  88. ;
  89. while (MailboxCacheRevolving.Pop(0))
  90. ;
  91. while (MailboxCacheHTSwap.Pop(0))
  92. ;
  93. while (MailboxCacheReadAsFilled.Pop(0))
  94. ;
  95. while (MailboxCacheTinyReadAsFilled.Pop(0))
  96. ;
  97. }
  98. bool TMailboxTable::Cleanup() {
  99. bool done = true;
  100. for (ui32 i = 0; i < LastAllocatedLine; ++i) {
  101. if (TMailboxLineHeader* lineHeader = Lines[i]) {
  102. switch (lineHeader->MailboxType) {
  103. case TMailboxType::Simple:
  104. done &= CleanupMailboxLine<TSimpleMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  105. break;
  106. case TMailboxType::Revolving:
  107. done &= CleanupMailboxLine<TRevolvingMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  108. break;
  109. case TMailboxType::HTSwap:
  110. done &= CleanupMailboxLine<THTSwapMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  111. break;
  112. case TMailboxType::ReadAsFilled:
  113. done &= CleanupMailboxLine<TReadAsFilledMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  114. break;
  115. case TMailboxType::TinyReadAsFilled:
  116. done &= CleanupMailboxLine<TTinyReadAsFilledMailbox>((ui8*)lineHeader + 64, (ui8*)lineHeader + LineSize);
  117. break;
  118. default:
  119. Y_FAIL();
  120. }
  121. }
  122. }
  123. return done;
  124. }
  125. TMailboxHeader* TMailboxTable::Get(ui32 hint) {
  126. // get line
  127. const ui32 lineIndex = (hint & LineIndexMask) >> LineIndexShift;
  128. const ui32 lineHint = hint & LineHintMask;
  129. Y_ABORT_UNLESS((lineIndex < MaxLines) && (lineHint < LineSize / 64));
  130. if (lineHint == 0)
  131. return nullptr;
  132. if (TMailboxLineHeader* const x = AtomicLoad(Lines + lineIndex)) {
  133. switch (x->MailboxType) {
  134. case TMailboxType::Simple:
  135. return TSimpleMailbox::Get(lineHint, x);
  136. case TMailboxType::Revolving:
  137. return TRevolvingMailbox::Get(lineHint, x);
  138. case TMailboxType::HTSwap:
  139. return THTSwapMailbox::Get(lineHint, x);
  140. case TMailboxType::ReadAsFilled:
  141. return TReadAsFilledMailbox::Get(lineHint, x);
  142. case TMailboxType::TinyReadAsFilled:
  143. return TTinyReadAsFilledMailbox::Get(lineHint, x);
  144. default:
  145. Y_VERIFY_DEBUG(false);
  146. break;
  147. }
  148. }
  149. return nullptr;
  150. }
  151. template <TMailboxTable::TEPScheduleActivationFunction EPSpecificScheduleActivation>
  152. bool TMailboxTable::GenericSendTo(TAutoPtr<IEventHandle>& ev, IExecutorPool* executorPool) {
  153. const TActorId& recipient = ev->GetRecipientRewrite();
  154. const ui32 hint = recipient.Hint();
  155. // copy-paste from Get to avoid duplicated type-switches
  156. const ui32 lineIndex = (hint & LineIndexMask) >> LineIndexShift;
  157. const ui32 lineHint = hint & LineHintMask;
  158. Y_ABORT_UNLESS((lineIndex < MaxLines) && (lineHint < LineSize / 64));
  159. if (lineHint == 0)
  160. return false;
  161. if (TMailboxLineHeader* const x = AtomicLoad(Lines + lineIndex)) {
  162. switch (x->MailboxType) {
  163. case TMailboxType::Simple: {
  164. TSimpleMailbox* const mailbox = TSimpleMailbox::Get(lineHint, x);
  165. mailbox->Push(recipient.LocalId());
  166. #if (!defined(_tsan_enabled_))
  167. Y_VERIFY_DEBUG(mailbox->Type == (ui32)x->MailboxType);
  168. #endif
  169. mailbox->Queue.Push(ev.Release());
  170. if (mailbox->MarkForSchedule()) {
  171. RelaxedStore<NHPTimer::STime>(&mailbox->ScheduleMoment, GetCycleCountFast());
  172. (executorPool->*EPSpecificScheduleActivation)(hint);
  173. }
  174. }
  175. return true;
  176. case TMailboxType::Revolving: {
  177. // The actorid could be stale and coming from a different machine. If local process has restarted than
  178. // the stale actorid coming from a remote machine might be referencing an actor with simple mailbox
  179. // which is smaller than revolving mailbox. In this cases 'lineHint' index might be greater than actual
  180. // array size. Normally its ok to store stale event to other actor's valid mailbox beacuse Receive will
  181. // compare receiver actor id and discard stale event. But in this case we should discard the event right away
  182. // instead of trying to enque it to a mailbox at invalid address.
  183. // NOTE: lineHint is 1-based
  184. static_assert(TSimpleMailbox::AlignedSize() <= TRevolvingMailbox::AlignedSize(),
  185. "We expect that one line can store more simple mailboxes than revolving mailboxes");
  186. if (lineHint > TRevolvingMailbox::MaxMailboxesInLine())
  187. return false;
  188. TRevolvingMailbox* const mailbox = TRevolvingMailbox::Get(lineHint, x);
  189. mailbox->Push(recipient.LocalId());
  190. #if (!defined(_tsan_enabled_))
  191. Y_VERIFY_DEBUG(mailbox->Type == (ui32)x->MailboxType);
  192. #endif
  193. mailbox->QueueWriter.Push(ev.Release());
  194. if (mailbox->MarkForSchedule()) {
  195. RelaxedStore<NHPTimer::STime>(&mailbox->ScheduleMoment, GetCycleCountFast());
  196. (executorPool->*EPSpecificScheduleActivation)(hint);
  197. }
  198. }
  199. return true;
  200. case TMailboxType::HTSwap: {
  201. THTSwapMailbox* const mailbox = THTSwapMailbox::Get(lineHint, x);
  202. mailbox->Push(recipient.LocalId());
  203. #if (!defined(_tsan_enabled_))
  204. Y_VERIFY_DEBUG(mailbox->Type == (ui32)x->MailboxType);
  205. #endif
  206. mailbox->Queue.Push(ev.Release());
  207. if (mailbox->MarkForSchedule()) {
  208. RelaxedStore<NHPTimer::STime>(&mailbox->ScheduleMoment, GetCycleCountFast());
  209. (executorPool->*EPSpecificScheduleActivation)(hint);
  210. }
  211. }
  212. return true;
  213. case TMailboxType::ReadAsFilled: {
  214. if (lineHint > TReadAsFilledMailbox::MaxMailboxesInLine())
  215. return false;
  216. TReadAsFilledMailbox* const mailbox = TReadAsFilledMailbox::Get(lineHint, x);
  217. mailbox->Push(recipient.LocalId());
  218. #if (!defined(_tsan_enabled_))
  219. Y_VERIFY_DEBUG(mailbox->Type == (ui32)x->MailboxType);
  220. #endif
  221. mailbox->Queue.Push(ev.Release());
  222. if (mailbox->MarkForSchedule()) {
  223. RelaxedStore<NHPTimer::STime>(&mailbox->ScheduleMoment, GetCycleCountFast());
  224. (executorPool->*EPSpecificScheduleActivation)(hint);
  225. }
  226. }
  227. return true;
  228. case TMailboxType::TinyReadAsFilled: {
  229. if (lineHint > TTinyReadAsFilledMailbox::MaxMailboxesInLine())
  230. return false;
  231. TTinyReadAsFilledMailbox* const mailbox = TTinyReadAsFilledMailbox::Get(lineHint, x);
  232. mailbox->Push(recipient.LocalId());
  233. #if (!defined(_tsan_enabled_))
  234. Y_VERIFY_DEBUG(mailbox->Type == (ui32)x->MailboxType);
  235. #endif
  236. mailbox->Queue.Push(ev.Release());
  237. if (mailbox->MarkForSchedule()) {
  238. RelaxedStore<NHPTimer::STime>(&mailbox->ScheduleMoment, GetCycleCountFast());
  239. (executorPool->*EPSpecificScheduleActivation)(hint);
  240. }
  241. }
  242. return true;
  243. default:
  244. Y_FAIL("unknown mailbox type");
  245. }
  246. }
  247. return false;
  248. }
  249. ui32 TMailboxTable::AllocateMailbox(TMailboxType::EType type, ui64 revolvingCounter) {
  250. ui32 x = TryAllocateMailbox(type, revolvingCounter);
  251. if (x == 0)
  252. x = AllocateNewLine(type);
  253. return x;
  254. }
  255. ui32 TMailboxTable::TryAllocateMailbox(TMailboxType::EType type, ui64 revolvingCounter) {
  256. switch (type) {
  257. case TMailboxType::Simple:
  258. do {
  259. if (ui32 ret = MailboxCacheSimple.Pop(revolvingCounter)) {
  260. AtomicDecrement(CachedSimpleMailboxes);
  261. return ret;
  262. }
  263. } while (AtomicGet(CachedSimpleMailboxes) > (MailboxCacheSimple.Concurrency * 512));
  264. return 0;
  265. case TMailboxType::Revolving:
  266. do {
  267. if (ui32 ret = MailboxCacheRevolving.Pop(revolvingCounter)) {
  268. AtomicDecrement(CachedRevolvingMailboxes);
  269. return ret;
  270. }
  271. } while (AtomicGet(CachedRevolvingMailboxes) > (MailboxCacheRevolving.Concurrency * 512));
  272. return 0;
  273. case TMailboxType::HTSwap:
  274. do {
  275. if (ui32 ret = MailboxCacheHTSwap.Pop(revolvingCounter)) {
  276. AtomicDecrement(CachedHTSwapMailboxes);
  277. return ret;
  278. }
  279. } while (AtomicGet(CachedHTSwapMailboxes) > (MailboxCacheHTSwap.Concurrency * 512));
  280. return 0;
  281. case TMailboxType::ReadAsFilled:
  282. do {
  283. if (ui32 ret = MailboxCacheReadAsFilled.Pop(revolvingCounter)) {
  284. AtomicDecrement(CachedReadAsFilledMailboxes);
  285. return ret;
  286. }
  287. } while (AtomicGet(CachedReadAsFilledMailboxes) > (MailboxCacheReadAsFilled.Concurrency * 512));
  288. return 0;
  289. case TMailboxType::TinyReadAsFilled:
  290. do {
  291. if (ui32 ret = MailboxCacheTinyReadAsFilled.Pop(revolvingCounter)) {
  292. AtomicDecrement(CachedTinyReadAsFilledMailboxes);
  293. return ret;
  294. }
  295. } while (AtomicGet(CachedTinyReadAsFilledMailboxes) > (MailboxCacheTinyReadAsFilled.Concurrency * 512));
  296. return 0;
  297. default:
  298. Y_FAIL("Unknown mailbox type");
  299. }
  300. }
  301. template
  302. bool TMailboxTable::GenericSendTo<&IExecutorPool::ScheduleActivation>(TAutoPtr<IEventHandle>& ev, IExecutorPool* executorPool);
  303. template
  304. bool TMailboxTable::GenericSendTo<&IExecutorPool::SpecificScheduleActivation>(TAutoPtr<IEventHandle>& ev, IExecutorPool* executorPool);
  305. void TMailboxTable::ReclaimMailbox(TMailboxType::EType type, ui32 hint, ui64 revolvingCounter) {
  306. if (hint != 0) {
  307. switch (type) {
  308. case TMailboxType::Simple:
  309. MailboxCacheSimple.Push(hint, revolvingCounter);
  310. AtomicIncrement(CachedSimpleMailboxes);
  311. break;
  312. case TMailboxType::Revolving:
  313. MailboxCacheRevolving.Push(hint, revolvingCounter);
  314. AtomicIncrement(CachedRevolvingMailboxes);
  315. break;
  316. case TMailboxType::HTSwap:
  317. MailboxCacheHTSwap.Push(hint, revolvingCounter);
  318. AtomicIncrement(CachedHTSwapMailboxes);
  319. break;
  320. case TMailboxType::ReadAsFilled:
  321. MailboxCacheReadAsFilled.Push(hint, revolvingCounter);
  322. AtomicIncrement(CachedReadAsFilledMailboxes);
  323. break;
  324. case TMailboxType::TinyReadAsFilled:
  325. MailboxCacheTinyReadAsFilled.Push(hint, revolvingCounter);
  326. AtomicIncrement(CachedTinyReadAsFilledMailboxes);
  327. break;
  328. default:
  329. Y_FAIL();
  330. }
  331. }
  332. }
  333. TMailboxHeader::TMailboxHeader(TMailboxType::EType type)
  334. : ExecutionState(TExecutionState::Free)
  335. , Reserved(0)
  336. , Type(type)
  337. , ActorPack(TMailboxActorPack::Simple)
  338. , Knobs(0)
  339. {
  340. ActorsInfo.Simple.ActorId = 0;
  341. ActorsInfo.Simple.Actor = nullptr;
  342. }
  343. TMailboxHeader::~TMailboxHeader() {
  344. CleanupActors();
  345. }
  346. bool TMailboxHeader::CleanupActors() {
  347. bool done = true;
  348. switch (ActorPack) {
  349. case TMailboxActorPack::Simple: {
  350. if (ActorsInfo.Simple.ActorId != 0) {
  351. delete ActorsInfo.Simple.Actor;
  352. done = false;
  353. }
  354. break;
  355. }
  356. case TMailboxActorPack::Map: {
  357. for (auto& [actorId, actor] : *ActorsInfo.Map.ActorsMap) {
  358. delete actor;
  359. }
  360. delete ActorsInfo.Map.ActorsMap;
  361. done = false;
  362. break;
  363. }
  364. case TMailboxActorPack::Array: {
  365. for (ui64 i = 0; i < ActorsInfo.Array.ActorsCount; ++i) {
  366. delete ActorsInfo.Array.ActorsArray->Actors[i].Actor;
  367. }
  368. delete ActorsInfo.Array.ActorsArray;
  369. done = false;
  370. break;
  371. }
  372. }
  373. ActorPack = TMailboxActorPack::Simple;
  374. ActorsInfo.Simple.ActorId = 0;
  375. ActorsInfo.Simple.Actor = nullptr;
  376. return done;
  377. }
  378. std::pair<ui32, ui32> TMailboxHeader::CountMailboxEvents(ui64 localActorId, ui32 maxTraverse) {
  379. switch (Type) {
  380. case TMailboxType::Simple:
  381. return static_cast<TMailboxTable::TSimpleMailbox*>(this)->CountSimpleMailboxEvents(localActorId, maxTraverse);
  382. case TMailboxType::Revolving:
  383. return static_cast<TMailboxTable::TRevolvingMailbox*>(this)->CountRevolvingMailboxEvents(localActorId, maxTraverse);
  384. default:
  385. return {0, 0};
  386. }
  387. }
  388. TMailboxUsageImpl<true>::~TMailboxUsageImpl() {
  389. while (auto *e = PendingEventQueue.Pop()) {
  390. delete e;
  391. }
  392. }
  393. void TMailboxUsageImpl<true>::Push(ui64 localId) {
  394. PendingEventQueue.Push(new TPendingEvent{localId, GetCycleCountFast()});
  395. }
  396. void TMailboxUsageImpl<true>::ProcessEvents(TMailboxHeader *mailbox) {
  397. while (std::unique_ptr<TPendingEvent> e{PendingEventQueue.Pop()}) {
  398. if (IActor *actor = mailbox->FindActor(e->LocalId)) {
  399. actor->OnEnqueueEvent(e->Timestamp);
  400. }
  401. }
  402. }
  403. TMailboxTable::TSimpleMailbox::TSimpleMailbox()
  404. : TMailboxHeader(TMailboxType::Simple)
  405. , ScheduleMoment(0)
  406. {
  407. }
  408. TMailboxTable::TSimpleMailbox::~TSimpleMailbox() {
  409. CleanupEvents();
  410. }
  411. bool TMailboxTable::TSimpleMailbox::CleanupEvents() {
  412. const bool done = (Queue.Head() == nullptr);
  413. while (IEventHandle* ev = Queue.Pop())
  414. delete ev;
  415. return done;
  416. }
  417. std::pair<ui32, ui32> TMailboxTable::TSimpleMailbox::CountSimpleMailboxEvents(ui64 localActorId, ui32 maxTraverse) {
  418. ui32 local = 0;
  419. ui32 total = 0;
  420. auto it = Queue.ReadIterator();
  421. while (IEventHandle* x = it.Next()) {
  422. ++total;
  423. if (x->GetRecipientRewrite().LocalId() == localActorId)
  424. ++local;
  425. if (total >= maxTraverse)
  426. break;
  427. }
  428. return std::make_pair(local, total);
  429. }
  430. TMailboxTable::TRevolvingMailbox::TRevolvingMailbox()
  431. : TMailboxHeader(TMailboxType::Revolving)
  432. , QueueWriter(QueueReader)
  433. , Reserved1(0)
  434. , Reserved2(0)
  435. , ScheduleMoment(0)
  436. {
  437. }
  438. TMailboxTable::TRevolvingMailbox::~TRevolvingMailbox() {
  439. CleanupEvents();
  440. }
  441. bool TMailboxTable::TRevolvingMailbox::CleanupEvents() {
  442. const bool done = (QueueReader.Head() == nullptr);
  443. while (IEventHandle* ev = QueueReader.Pop())
  444. delete ev;
  445. return done;
  446. }
  447. std::pair<ui32, ui32> TMailboxTable::TRevolvingMailbox::CountRevolvingMailboxEvents(ui64 localActorId, ui32 maxTraverse) {
  448. ui32 local = 0;
  449. ui32 total = 0;
  450. auto it = QueueReader.Iterator();
  451. while (IEventHandle* x = it.Next()) {
  452. ++total;
  453. if (x->GetRecipientRewrite().LocalId() == localActorId)
  454. ++local;
  455. if (total >= maxTraverse)
  456. break;
  457. }
  458. return std::make_pair(local, total);
  459. }
  460. template <typename T>
  461. static ui32 InitNewLine(ui8* x, ui8* end) {
  462. const ui32 sx = T::AlignedSize();
  463. for (ui32 index = 1; x + sx <= end; x += sx, ++index)
  464. ::new (x) T();
  465. return sx;
  466. }
  467. ui32 TMailboxTable::AllocateNewLine(TMailboxType::EType type) {
  468. ui8* ptr = (ui8*)malloc(LineSize);
  469. ui8* end = ptr + LineSize;
  470. const ui32 lineIndex = (ui32)AtomicIncrement(LastAllocatedLine) - 1;
  471. const ui32 lineIndexMask = (lineIndex << LineIndexShift) & LineIndexMask;
  472. // first 64 bytes is TMailboxLineHeader
  473. TMailboxLineHeader* header = ::new (ptr) TMailboxLineHeader(type, lineIndex);
  474. ui8* x = ptr + 64;
  475. ui32 sx = 0;
  476. TMailboxCache* cache = nullptr;
  477. TAtomic* counter = nullptr;
  478. switch (type) {
  479. case TMailboxType::Simple:
  480. sx = InitNewLine<TSimpleMailbox>(x, end);
  481. cache = &MailboxCacheSimple;
  482. counter = &CachedSimpleMailboxes;
  483. break;
  484. case TMailboxType::Revolving:
  485. sx = InitNewLine<TRevolvingMailbox>(x, end);
  486. cache = &MailboxCacheRevolving;
  487. counter = &CachedRevolvingMailboxes;
  488. break;
  489. case TMailboxType::HTSwap:
  490. sx = InitNewLine<THTSwapMailbox>(x, end);
  491. cache = &MailboxCacheHTSwap;
  492. counter = &CachedHTSwapMailboxes;
  493. break;
  494. case TMailboxType::ReadAsFilled:
  495. sx = InitNewLine<TReadAsFilledMailbox>(x, end);
  496. cache = &MailboxCacheReadAsFilled;
  497. counter = &CachedReadAsFilledMailboxes;
  498. break;
  499. case TMailboxType::TinyReadAsFilled:
  500. sx = InitNewLine<TTinyReadAsFilledMailbox>(x, end);
  501. cache = &MailboxCacheTinyReadAsFilled;
  502. counter = &CachedTinyReadAsFilledMailboxes;
  503. break;
  504. default:
  505. Y_FAIL();
  506. }
  507. AtomicStore(Lines + lineIndex, header);
  508. ui32 ret = lineIndexMask | 1;
  509. ui32 index = 2;
  510. for (ui32 endIndex = LineSize / sx; index != endIndex;) {
  511. const ui32 bufSize = 8;
  512. ui32 buf[bufSize];
  513. ui32 bufIndex;
  514. for (bufIndex = 0; index != endIndex && bufIndex != bufSize; ++bufIndex, ++index)
  515. buf[bufIndex] = lineIndexMask | index;
  516. cache->PushBulk(buf, bufIndex, index);
  517. AtomicAdd(*counter, bufIndex);
  518. }
  519. AtomicAdd(AllocatedMailboxCount, index - 1);
  520. return ret;
  521. }
  522. bool TMailboxTable::SendTo(TAutoPtr<IEventHandle>& ev, IExecutorPool* executorPool) {
  523. return GenericSendTo<&IExecutorPool::ScheduleActivation>(ev, executorPool);
  524. }
  525. bool TMailboxTable::SpecificSendTo(TAutoPtr<IEventHandle>& ev, IExecutorPool* executorPool) {
  526. return GenericSendTo<&IExecutorPool::SpecificScheduleActivation>(ev, executorPool);
  527. }
  528. }