Skip to content

Commit

Permalink
Fixes in release phase
Browse files Browse the repository at this point in the history
  • Loading branch information
@vishalgupta97
vishalgupta97 committed Aug 23, 2024
1 parent 03ee71e commit 3a43dfc
Showing 1 changed file with 76 additions and 109 deletions.
185 changes: 76 additions & 109 deletions src/rt/sched/behaviourcore.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -923,58 +923,55 @@ namespace verona::rt
return;
}

if (is_read_only())
if(next_slot() == nullptr)
{
if (next_slot() == nullptr)
auto slot_addr = this;
if (cown()->last_slot.compare_exchange_strong(
slot_addr, nullptr, std::memory_order_acq_rel))
{
auto slot_addr = this;
if (cown()->last_slot.compare_exchange_strong(
slot_addr, nullptr, std::memory_order_acq_rel))
if(is_read_only() && cown()->read_ref_count.release_read())
{
Logging::cout() << *this << " Reader releasing the cown "
<< Logging::endl;
if (cown()->read_ref_count.release_read())
Logging::cout()
<< *this << "Last Reader releasing the cown " << Logging::endl;
// Last reader
auto w =
cown()->next_writer.exchange(nullptr, std::memory_order_acq_rel);
if (w != nullptr)
{
Logging::cout()
<< *this << "Last Reader releasing the cown " << Logging::endl;
// Last reader
auto w =
cown()->next_writer.exchange(nullptr, std::memory_order_acq_rel);
if (w != nullptr)
{
Logging::cout()
<< *this << " Last Reader waking up next writer cown "
<< " writer " << *w << Logging::endl;
w->resolve();
}
yield();

// Release cown as this will be set by the new thread joining the
// queue.
Logging::cout() << *this << " Last reader No more work for cown "
<< Logging::endl;
shared::release(ThreadAlloc::get(), cown());
<< *this << " Last Reader waking up next writer cown "
<< " writer " << *w << Logging::endl;
w->resolve();
}
yield();

// Release cown as this will be set by the new thread joining the
// queue.
Logging::cout() << *this << " CAS Success No more work for cown "
Logging::cout() << *this << " Last reader No more work for cown "
<< Logging::endl;
shared::release(ThreadAlloc::get(), cown());
return;
}
Logging::cout() << *this
<< " Reader another thread extending the chain cown "

yield();
// Release cown as this will be set by the new thread joining the
// queue.
Logging::cout() << *this << " CAS Success No more work for cown "
<< Logging::endl;
// If we failed, then the another thread is extending the chain
while (next_slot() == nullptr)
{
Systematic::yield_until(
[this]() { return (next_slot() != nullptr); });
Aal::pause();
}
shared::release(ThreadAlloc::get(), cown());
return;
}

// If we failed, then the another thread is extending the chain
while (next_slot() == nullptr)
{
Systematic::yield_until(
[this]() { return (next_slot() != nullptr); });
Aal::pause();
}
}

if(is_read_only())
{
if (is_next_slot_writer())
{
Logging::cout() << *this << "Reader setting next writer variable "
Expand Down Expand Up @@ -1004,82 +1001,52 @@ namespace verona::rt
<< Logging::endl;
shared::release(ThreadAlloc::get(), cown());
}
return;
}

if(is_next_slot_writer()) {
Logging::cout() << *this
<< " Writer waking up next writer cown next slot "
<< *next_behaviour() << Logging::endl;
next_behaviour()->resolve();
return;
}
else
{
if (next_slot() == nullptr)
{
auto slot_addr = this;
if (cown()->last_slot.compare_exchange_strong(
slot_addr, nullptr, std::memory_order_acq_rel))
{
yield();
Logging::cout() << *this
<< " No more work Last writer releasing the cown "
<< Logging::endl;
shared::release(ThreadAlloc::get(), cown());
return;
}
// If we failed, then the another thread is extending the chain
while (next_slot() == nullptr)
{
Systematic::yield_until(
[this]() { return (next_slot() != nullptr); });
Aal::pause();
}
}

if (is_next_slot_read_only())
{
std::vector<Slot*> reader_queue;
yield();
bool first_reader = cown()->read_ref_count.add_read();
Logging::cout() << *this
<< " Writer waking up next reader and acquiring "
"reference count for first reader. next slot "
<< *next_slot() << Logging::endl;
assert(first_reader);
yield();
Cown::acquire(cown());
yield();
next_slot()->set_active();
yield();
reader_queue.push_back(next_slot());

auto curr_slot = next_slot();
while (curr_slot->is_next_slot_read_only())
{
yield();
assert(curr_slot->next_slot() != nullptr);
auto reader = curr_slot->next_slot();
yield();
reader->set_active();
Logging::cout()
<< *this << " Writer loop waking up next reader cown. Next slot "
<< *curr_slot->next_slot() << Logging::endl;
reader_queue.push_back(reader);
yield();
curr_slot = reader;
}

// Add read count for readers. First reader is already added in rcount
cown()->read_ref_count.add_read(
reader_queue.size() - 1);
std::vector<Slot*> reader_queue;
bool first_reader = cown()->read_ref_count.add_read();
Logging::cout() << *this
<< " Writer waking up next reader and acquiring "
"reference count for first reader. next slot "
<< *next_slot() << Logging::endl;
assert(first_reader);
Cown::acquire(cown());
yield();
next_slot()->set_active();
reader_queue.push_back(next_slot());

auto curr_slot = next_slot();
while (curr_slot->is_next_slot_read_only())
{
yield();
assert(curr_slot->next_slot() != nullptr);
auto reader = curr_slot->next_slot();
yield();
reader->set_active();
Logging::cout()
<< *this << " Writer loop waking up next reader cown. Next slot "
<< *curr_slot->next_slot() << Logging::endl;
reader_queue.push_back(reader);
yield();
curr_slot = reader;
}

yield();
// Add read count for readers. First reader is already added in rcount
cown()->read_ref_count.add_read(reader_queue.size() - 1);

for (auto reader : reader_queue)
{
reader->get_behaviour()->resolve(1, false);
}
}
else
{
Logging::cout() << *this
<< " Writer waking up next writer cown next slot "
<< *next_behaviour() << Logging::endl;
next_behaviour()->resolve();
}
for (auto reader : reader_queue)
{
reader->get_behaviour()->resolve(1, false);
yield();
}
}
} // namespace verona::rt

0 comments on commit 3a43dfc

Please sign in to comment.