diff --git a/simgear/emesary/INotification.hxx b/simgear/emesary/INotification.hxx index 0d6a517a..648e1b39 100644 --- a/simgear/emesary/INotification.hxx +++ b/simgear/emesary/INotification.hxx @@ -21,14 +21,20 @@ * Copyright (C)2019 Richard Harrison Licenced under GPL2 or later. * *---------------------------------------------------------------------------*/ +#include + namespace simgear { namespace Emesary { /// Interface (base class) for all notifications. - class INotification + class INotification: public SGReferenced { public: + virtual ~INotification() + { + + } // text representation of notification type. must be unique across all notifications virtual const char *GetType() = 0; @@ -49,6 +55,7 @@ namespace simgear /// is returned as the status. virtual bool IsComplete() { return true; } }; + typedef SGSharedPtr INotificationPtr; } } #endif diff --git a/simgear/emesary/IReceiver.hxx b/simgear/emesary/IReceiver.hxx index 020f293e..b5197517 100644 --- a/simgear/emesary/IReceiver.hxx +++ b/simgear/emesary/IReceiver.hxx @@ -19,33 +19,31 @@ * Copyright (C)2019 Richard Harrison Licenced under GPL2 or later. * *---------------------------------------------------------------------------*/ - -#include - namespace simgear { namespace Emesary { /// Interface (base class) for a recipeint. - class IReceiver : public SGReferenced + class IReceiver { public: virtual ~IReceiver() = default; /// Receive notification - must be implemented - virtual ReceiptStatus Receive(INotification& message) = 0; + virtual ReceiptStatus Receive(INotificationPtr message) = 0; /// Called when registered at a transmitter - virtual void OnRegisteredAtTransmitter(class Transmitter* p) + virtual void OnRegisteredAtTransmitter(class Transmitter *p) { } /// Called when de-registered at a transmitter - virtual void OnDeRegisteredAtTransmitter(class Transmitter* p) + virtual void OnDeRegisteredAtTransmitter(class Transmitter *p) { } }; + typedef IReceiver* IReceiverPtr; } } diff --git a/simgear/emesary/ITransmitter.hxx b/simgear/emesary/ITransmitter.hxx index b3418534..118a5324 100644 --- a/simgear/emesary/ITransmitter.hxx +++ b/simgear/emesary/ITransmitter.hxx @@ -21,12 +21,11 @@ *---------------------------------------------------------------------------*/ #include -#include + namespace simgear { namespace Emesary { - typedef SGSharedPtr IReceiverPtr; /// Interface (base clasee) for a transmitter. /// Transmits Message derived objects. Each instance of this class provides a /// event/databus to which any number of receivers can attach to. @@ -34,9 +33,9 @@ namespace simgear { public: // Registers a recipient to receive message from this transmitter - virtual void Register(IReceiverPtr R) = 0; + virtual void Register(IReceiverPtr R) = 0; // Removes a recipient from from this transmitter - virtual void DeRegister(IReceiverPtr R) = 0; + virtual void DeRegister(IReceiverPtr R) = 0; //Notify all registered recipients. Stop when receipt status of abort or finished are received. @@ -45,10 +44,10 @@ namespace simgear // - Fail > message not handled. A status of Abort from a recipient will result in our status // being fail as Abort means that the message was not and cannot be handled, and // allows for usages such as access controls. - virtual ReceiptStatus NotifyAll(INotification& M) = 0; + virtual ReceiptStatus NotifyAll(INotificationPtr M) = 0; /// number of recipients - virtual size_t Count() const = 0; + virtual size_t Count() const = 0; }; } } diff --git a/simgear/emesary/Transmitter.hxx b/simgear/emesary/Transmitter.hxx index 1cd1beaa..d51fe7ed 100644 --- a/simgear/emesary/Transmitter.hxx +++ b/simgear/emesary/Transmitter.hxx @@ -28,231 +28,232 @@ #include #include #include +#include +#include "ITransmitter.hxx" namespace simgear { - namespace Emesary - { - // Implementation of a ITransmitter - class Transmitter : public ITransmitter - { - protected: + namespace Emesary + { + // Implementation of a ITransmitter + class Transmitter : public ITransmitter + { + protected: typedef std::vector RecipientList; - RecipientList recipient_list; - RecipientList new_recipient_list; - RecipientList deleted_recipient_list; + RecipientList recipient_list; + RecipientList new_recipient_list; + RecipientList deleted_recipient_list; - std::mutex _lock; - std::atomic receiveDepth; - std::atomic sentMessageCount; - std::atomic recipientCount; - std::atomic pendingDeletions; - std::atomic pendingAdditions; + std::mutex _lock; + std::atomic receiveDepth; + std::atomic sentMessageCount; + std::atomic recipientCount; + std::atomic pendingDeletions; + std::atomic pendingAdditions; - public: - Transmitter() : - receiveDepth(0), - sentMessageCount(0), - recipientCount(0), - pendingDeletions(0), - pendingAdditions(0) - { - } - - virtual ~Transmitter() - { - } - - // Registers an object to receive messsages from this transmitter. - // This object is added to the top of the list of objects to be notified. This is deliberate as - // the sequence of registration and message receipt can influence the way messages are processing - // when ReceiptStatus of Abort or Finished are encountered. So it was a deliberate decision that the - // most recently registered recipients should process the messages/events first. - virtual void Register(IReceiverPtr r) - { - std::lock_guard scopeLock(_lock); - - RecipientList::iterator deleted_location = std::find(deleted_recipient_list.begin(), deleted_recipient_list.end(), r); - if (deleted_location != deleted_recipient_list.end()) - deleted_recipient_list.erase(deleted_location); - - RecipientList::iterator location = std::find(recipient_list.begin(), recipient_list.end(), r); - if (location == recipient_list.end()) + public: + Transmitter() : + receiveDepth(0), + sentMessageCount(0), + recipientCount(0), + pendingDeletions(0), + pendingAdditions(0) { - RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r); - if (location == new_recipient_list.end()) { - new_recipient_list.insert(new_recipient_list.begin(), r); - pendingAdditions++; + } + + virtual ~Transmitter() + { + } + + // Registers an object to receive messsages from this transmitter. + // This object is added to the top of the list of objects to be notified. This is deliberate as + // the sequence of registration and message receipt can influence the way messages are processing + // when ReceiptStatus of Abort or Finished are encountered. So it was a deliberate decision that the + // most recently registered recipients should process the messages/events first. + virtual void Register(IReceiverPtr r) + { + std::lock_guard scopeLock(_lock); + + RecipientList::iterator deleted_location = std::find(deleted_recipient_list.begin(), deleted_recipient_list.end(), r); + if (deleted_location != deleted_recipient_list.end()) + deleted_recipient_list.erase(deleted_location); + + RecipientList::iterator location = std::find(recipient_list.begin(), recipient_list.end(), r); + if (location == recipient_list.end()) + { + RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r); + if (location == new_recipient_list.end()) { + new_recipient_list.insert(new_recipient_list.begin(), r); + pendingAdditions++; + } } } - } - // Removes an object from receving message from this transmitter - virtual void DeRegister(IReceiverPtr r) - { - std::lock_guard scopeLock(_lock); + // Removes an object from receving message from this transmitter + virtual void DeRegister(IReceiverPtr r) + { + std::lock_guard scopeLock(_lock); - if (new_recipient_list.size()) - { - RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r); + if (new_recipient_list.size()) + { + RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r); - if (location != new_recipient_list.end()) - new_recipient_list.erase(location); + if (location != new_recipient_list.end()) + new_recipient_list.erase(location); + } + deleted_recipient_list.push_back(r); + pendingDeletions++; } - deleted_recipient_list.push_back(r); - pendingDeletions++; - } - // this will purge the recipients that are marked as deleted - // it will only do this when the receive depth is zero - i.e. this - // notification is being sent out from outside a recipient notify.(because it is - // fine for a recipient to retransmit another notification as a result of receiving a notification) - // - // also we can quickly check to see if we have any pending deletions before doing anything. - void AddRemoveIFAppropriate() - { - std::lock_guard scopeLock(_lock); + // this will purge the recipients that are marked as deleted + // it will only do this when the receive depth is zero - i.e. this + // notification is being sent out from outside a recipient notify.(because it is + // fine for a recipient to retransmit another notification as a result of receiving a notification) + // + // also we can quickly check to see if we have any pending deletions before doing anything. + void AddRemoveIFAppropriate() + { + std::lock_guard scopeLock(_lock); - /// handle pending deletions first. - if (pendingDeletions > 0) { + /// handle pending deletions first. + if (pendingDeletions > 0) { - /// - /// remove deleted recipients from the main list. - std::for_each(deleted_recipient_list.begin(), deleted_recipient_list.end(), - [this](IReceiverPtr r) { + /// + /// remove deleted recipients from the main list. + std::for_each(deleted_recipient_list.begin(), deleted_recipient_list.end(), + [this](IReceiverPtr r) { RecipientList::iterator location = std::find(recipient_list.begin(), recipient_list.end(), r); if (location != recipient_list.end()) { r->OnDeRegisteredAtTransmitter(this); recipient_list.erase(location); - } + } }); - recipientCount -= pendingDeletions; - deleted_recipient_list.erase(deleted_recipient_list.begin(), deleted_recipient_list.end()); - pendingDeletions = 0; // can do this because we are guarded - } + recipientCount -= pendingDeletions; + deleted_recipient_list.erase(deleted_recipient_list.begin(), deleted_recipient_list.end()); + pendingDeletions = 0; // can do this because we are guarded + } - if (pendingAdditions) { - /// firstly remove items from the new list that are already in the list - std::for_each(recipient_list.begin(), recipient_list.end(), - [this](IReceiverPtr r) { + if (pendingAdditions) { + /// firstly remove items from the new list that are already in the list + std::for_each(recipient_list.begin(), recipient_list.end(), + [this](IReceiverPtr r) { RecipientList::iterator location = std::find(new_recipient_list.begin(), new_recipient_list.end(), r); if (location != new_recipient_list.end()) { new_recipient_list.erase(location); pendingAdditions--; - } + } }); - std::for_each(new_recipient_list.begin(), new_recipient_list.end(), - [this](IReceiverPtr r) { + std::for_each(new_recipient_list.begin(), new_recipient_list.end(), + [this](IReceiverPtr r) { r->OnRegisteredAtTransmitter(this); }); - recipient_list.insert(recipient_list.begin(), - std::make_move_iterator(new_recipient_list.begin()), - std::make_move_iterator(new_recipient_list.end())); + recipient_list.insert(recipient_list.begin(), + std::make_move_iterator(new_recipient_list.begin()), + std::make_move_iterator(new_recipient_list.end())); - new_recipient_list.erase(new_recipient_list.begin(), new_recipient_list.end()); - recipientCount += pendingAdditions; - pendingAdditions = 0; + new_recipient_list.erase(new_recipient_list.begin(), new_recipient_list.end()); + recipientCount += pendingAdditions; + pendingAdditions = 0; + } } - } - // Notify all registered recipients. Stop when receipt status of abort or finished are received. - // The receipt status from this method will be - // - OK > message handled - // - Fail > message not handled. A status of Abort from a recipient will result in our status - // being fail as Abort means that the message was not and cannot be handled, and - // allows for usages such as access controls. - virtual ReceiptStatus NotifyAll(INotification& M) - { - ReceiptStatus return_status = ReceiptStatus::NotProcessed; + // Notify all registered recipients. Stop when receipt status of abort or finished are received. + // The receipt status from this method will be + // - OK > message handled + // - Fail > message not handled. A status of Abort from a recipient will result in our status + // being fail as Abort means that the message was not and cannot be handled, and + // allows for usages such as access controls. + virtual ReceiptStatus NotifyAll(INotificationPtr M) + { + ReceiptStatus return_status = ReceiptStatus::NotProcessed; - auto v = receiveDepth.fetch_add(1, std::memory_order_relaxed); - if (v == 0) - AddRemoveIFAppropriate(); + auto v = receiveDepth.fetch_add(1, std::memory_order_relaxed); + if (v == 0) + AddRemoveIFAppropriate(); - sentMessageCount++; + sentMessageCount++; - bool finished = false; + bool finished = false; - size_t idx = 0; - do { + size_t idx = 0; + do { - if (idx < recipient_list.size()) { - IReceiverPtr R = recipient_list[idx++]; + if (idx < recipient_list.size()) { + IReceiverPtr R = recipient_list[idx++]; - if (R != nullptr) - { - Emesary::ReceiptStatus rstat = R->Receive(M); - switch (rstat) - { - case ReceiptStatus::Fail: - return_status = ReceiptStatus::Fail; - break; + if (R != nullptr) + { + Emesary::ReceiptStatus rstat = R->Receive(M); + switch (rstat) + { + case ReceiptStatus::Fail: + return_status = ReceiptStatus::Fail; + break; - case ReceiptStatus::Pending: - return_status = ReceiptStatus::Pending; - break; + case ReceiptStatus::Pending: + return_status = ReceiptStatus::Pending; + break; - case ReceiptStatus::PendingFinished: - return_status = rstat; - finished = true; - break; + case ReceiptStatus::PendingFinished: + return_status = rstat; + finished = true; + break; - case ReceiptStatus::NotProcessed: - break; + case ReceiptStatus::NotProcessed: + break; - case ReceiptStatus::OK: - if (return_status == ReceiptStatus::NotProcessed) - return_status = rstat; - break; + case ReceiptStatus::OK: + if (return_status == ReceiptStatus::NotProcessed) + return_status = rstat; + break; - case ReceiptStatus::Abort: - finished = true; - return_status = ReceiptStatus::Abort; - break; + case ReceiptStatus::Abort: + finished = true; + return_status = ReceiptStatus::Abort; + break; - case ReceiptStatus::Finished: - finished = true; - return_status = ReceiptStatus::OK;; - break; - } - } + case ReceiptStatus::Finished: + finished = true; + return_status = ReceiptStatus::OK;; + break; + } + } - } - else - break; - } while (!finished); + } + else + break; + } while (!finished); + receiveDepth--; + return return_status; + } - receiveDepth--; - return return_status; - } + // number of sent messages. + size_t SentMessageCount() const + { + return sentMessageCount; + } - // number of sent messages. - int SentMessageCount() const - { - return sentMessageCount; - } - - // number of currently registered recipients - // better to avoid using the size members on the list directly as - // using the atomics is lock free and threadsafe. - virtual size_t Count() const - { - return (recipientCount + pendingAdditions) - pendingDeletions; - } - // ascertain if a receipt status can be interpreted as failure. - static bool Failed(ReceiptStatus receiptStatus) - { - // - // failed is either Fail or Abort. - // NotProcessed isn't a failure because it hasn't been processed. - return receiptStatus == ReceiptStatus::Fail - || receiptStatus == ReceiptStatus::Abort; - } - }; - } + // number of currently registered recipients + // better to avoid using the size members on the list directly as + // using the atomics is lock free and threadsafe. + virtual size_t Count() const + { + return (recipientCount + pendingAdditions) - pendingDeletions; + } + // ascertain if a receipt status can be interpreted as failure. + static bool Failed(ReceiptStatus receiptStatus) + { + // + // failed is either Fail or Abort. + // NotProcessed isn't a failure because it hasn't been processed. + return receiptStatus == ReceiptStatus::Fail + || receiptStatus == ReceiptStatus::Abort; + } + }; + } } #endif diff --git a/simgear/emesary/notifications.hxx b/simgear/emesary/notifications.hxx index 0c98e3c6..94f8aeef 100644 --- a/simgear/emesary/notifications.hxx +++ b/simgear/emesary/notifications.hxx @@ -29,7 +29,7 @@ namespace simgear class MainLoopNotification : public simgear::Emesary::INotification { public: - enum Type { Started, Stopped, Begin, End }; + enum class Type { Started, Stopped, Begin, End }; MainLoopNotification(Type v) : _type(v) {} virtual Type GetValue() { return _type; } diff --git a/simgear/emesary/test_emesary.cxx b/simgear/emesary/test_emesary.cxx index 0ab974aa..9bcb81f6 100644 --- a/simgear/emesary/test_emesary.cxx +++ b/simgear/emesary/test_emesary.cxx @@ -5,11 +5,12 @@ #include #include -#include #include #include #include +#include +#include using std::cout; using std::cerr; @@ -17,7 +18,18 @@ using std::endl; std::atomic nthread {0}; std::atomic noperations {0}; -const int MaxIterations = 9999; +const int MaxIterationsBase = 9999999; +const int MaxIterationsThreaded = 90; +const int MaxIterationsBaseMultipleRecipients = 9999999; +const int num_threads = 220; + + +void summary(const SGTimeStamp& timeStamp, simgear::Emesary::Transmitter* transmitter, const char* id) { + printf("[%s]: invocations %zu\n", id, transmitter->SentMessageCount()); + double elapsed_seconds = timeStamp.elapsedMSec() / 1000.0f; + printf("[%s]: -> elapsed %d\n", id, timeStamp.elapsedMSec()); + printf("[%s]: took %lf seconds which is %lf/sec\n", id, elapsed_seconds, transmitter->SentMessageCount() / elapsed_seconds); +} class TestThreadNotification : public simgear::Emesary::INotification { @@ -29,51 +41,76 @@ public: virtual const char* GetType () { return baseValue; } }; -class TestThreadRecipient : public simgear::Emesary::IReceiver +class TestThreadBaseRecipient : public simgear::Emesary::IReceiver { public: - TestThreadRecipient() : receiveCount(0) + virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotificationPtr n) { + return simgear::Emesary::ReceiptStatus::NotProcessed; } +}; +class TestThreadRecipient : public simgear::Emesary::IReceiver +{ + simgear::Emesary::ITransmitter* transmitter; +public: + TestThreadRecipient(simgear::Emesary::ITransmitter* _transmitter, bool addDuringReceive) + : transmitter(_transmitter), addDuringReceive(addDuringReceive), receiveCount(0), ourType("TestThread") + { + r1 = new TestThreadBaseRecipient(); + } + simgear::Emesary::IReceiver *r1; + std::string ourType; + bool addDuringReceive; std::atomic receiveCount; - virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotification &n) + virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotificationPtr n) { - if (n.GetType() == (const char*)this) + if (ourType == n->GetType()) { +// SGSharedPtr r1 = new TestThreadBaseRecipient(); + // Unused: TestThreadNotification *tn = dynamic_cast(&n); receiveCount++; - TestThreadNotification onwardNotification("AL"); - simgear::Emesary::GlobalTransmitter::instance()->NotifyAll(onwardNotification); - + SGSharedPtr onwardNotification(new TestThreadNotification("AL")); + transmitter->NotifyAll(onwardNotification); + if (addDuringReceive) { + transmitter->Register(r1); + transmitter->NotifyAll(onwardNotification); + transmitter->DeRegister(r1); + } return simgear::Emesary::ReceiptStatus::OK; } - return simgear::Emesary::ReceiptStatus::OK; } }; class EmesaryTestThread : public SGThread { +public: + EmesaryTestThread(simgear::Emesary::ITransmitter* transmitter, bool _addDuringReceive) : addDuringReceive(_addDuringReceive), transmitter(transmitter) { + + } protected: + simgear::Emesary::ITransmitter* transmitter; + bool addDuringReceive; virtual void run() { int threadId = nthread.fetch_add(1); //System.Threading.Interlocked.Increment(ref nthread); //var rng = new Random(); - SGSharedPtr r = new TestThreadRecipient; + TestThreadRecipient *r = new TestThreadRecipient(transmitter, addDuringReceive); char temp[100]; sprintf(temp, "Notif %d", threadId); printf("starting thread %s\n", temp); - TestThreadNotification tn((const char*)&r); - for (int i = 0; i < MaxIterations; i++) + SGSharedPtr tn(new TestThreadNotification("TestThread")); + + for (int i = 0; i < MaxIterationsThreaded; i++) { - simgear::Emesary::IReceiverPtr ir(r); - simgear::Emesary::GlobalTransmitter::instance()->Register(ir); - simgear::Emesary::GlobalTransmitter::instance()->NotifyAll(tn); - simgear::Emesary::GlobalTransmitter::instance()->DeRegister(ir); + transmitter->Register(r); + transmitter->NotifyAll(tn); + transmitter->DeRegister(r); //System.Threading.Thread.Sleep(rng.Next(MaxSleep)); noperations++; } @@ -86,14 +123,13 @@ class EmesaryTest { public: - void Emesary_MultiThreadTransmitterTest() + void Emesary_MultiThreadTransmitterTest(simgear::Emesary::ITransmitter *transmitter, bool addDuringReceive) { - int num_threads = 12; std::list threads; for (int i = 0; i < num_threads; i++) { - EmesaryTestThread *thread = new EmesaryTestThread(); + EmesaryTestThread *thread = new EmesaryTestThread(transmitter, addDuringReceive); threads.push_back(thread); thread->start(); } @@ -106,28 +142,211 @@ public: void testEmesaryThreaded() { - SGSharedPtr r = new TestThreadRecipient; - simgear::Emesary::IReceiverPtr ir(r); - TestThreadNotification tn((const char*)r.ptr()); - simgear::Emesary::GlobalTransmitter::instance()->Register(ir); - for (int i = 0; i < MaxIterations*MaxIterations; i++) + printf("Testing multithreaded operations\n"); + + simgear::Emesary::Transmitter* globalTransmitter = simgear::Emesary::GlobalTransmitter::instance(); + + TestThreadRecipient *r = new TestThreadRecipient(globalTransmitter, false); + SGSharedPtr < TestThreadNotification> tn (new TestThreadNotification("TestThread")); + globalTransmitter->Register(r); + SGTimeStamp timeStamp; + timeStamp.stamp(); + printf(" -- simple receive\n"); + for (int i = 0; i < MaxIterationsThreaded; i++) { - simgear::Emesary::GlobalTransmitter::instance()->NotifyAll(tn); + globalTransmitter->NotifyAll(tn); //System.Threading.Thread.Sleep(rng.Next(MaxSleep)); noperations++; } - simgear::Emesary::GlobalTransmitter::instance()->DeRegister(ir); - printf("invocations %d\n", simgear::Emesary::GlobalTransmitter::instance()->SentMessageCount()); + + globalTransmitter->DeRegister(r); + printf("invocations %zu\n", globalTransmitter->SentMessageCount()); + double elapsed_seconds = timeStamp.elapsedMSec() / 1000.0f; + printf(" -> elapsed %d\n", timeStamp.elapsedMSec()); + printf("took %lf seconds which is %lf/sec\n", elapsed_seconds , globalTransmitter->SentMessageCount() / elapsed_seconds); EmesaryTest t; - t.Emesary_MultiThreadTransmitterTest(); + t.Emesary_MultiThreadTransmitterTest(globalTransmitter, false); + + elapsed_seconds = timeStamp.elapsedMSec() / 1000.0f; + printf(" -> elapsed %d\n", timeStamp.elapsedMSec()); + printf("took %lf seconds which is %lf/sec\n", elapsed_seconds, globalTransmitter->SentMessageCount() / elapsed_seconds); } +void testEmesaryThreadedAddDuringReceive() +{ + SGTimeStamp timeStamp; + timeStamp.stamp(); + simgear::Emesary::Transmitter* globalTransmitter = simgear::Emesary::GlobalTransmitter::instance(); + + EmesaryTest t; + t.Emesary_MultiThreadTransmitterTest(globalTransmitter, true); + summary(timeStamp, globalTransmitter, "ThreadedAddReceive"); +} +////////////////////// +/// basic tests + +class TestBaseNotification : public simgear::Emesary::INotification +{ +public: + TestBaseNotification() : index(0) {} + int index; + virtual const char* GetType() { return "Test"; } +}; + +static size_t TestDerefNotification_destructor_count = 0; +class TestDerefNotification : public simgear::Emesary::INotification +{ +public: + TestDerefNotification() {} + ~TestDerefNotification() override { + TestDerefNotification_destructor_count ++; + } + + virtual const char* GetType() { return "TestDerefNotification"; } +}; + +class TestBaseRecipient : public simgear::Emesary::IReceiver +{ +public: + TestBaseRecipient() : receiveCount(0) + { + } + + std::atomic receiveCount; + + virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotificationPtr n) + { + if (n->GetType() == "Test") + { + auto tbn = dynamic_pointer_cast(n); + SG_CHECK_EQUAL(receiveCount, tbn->index); + receiveCount++; + return simgear::Emesary::ReceiptStatus::OK; + } + + return simgear::Emesary::ReceiptStatus::OK; + } +}; + + +void testEmesaryBase() +{ + printf("Testing base functions\n"); + + simgear::Emesary::Transmitter* globalTransmitter = simgear::Emesary::GlobalTransmitter::instance(); + SGTimeStamp timeStamp; + timeStamp.stamp(); + + TestBaseRecipient *r = new TestBaseRecipient(); + + globalTransmitter->Register(r); + SG_CHECK_EQUAL(globalTransmitter->Count(), 1); + for (int i = 0; i < MaxIterationsBase; i++) + { + SGSharedPtr tn (new TestBaseNotification()); + tn->index = i; + globalTransmitter->NotifyAll(tn); + //System.Threading.Thread.Sleep(rng.Next(MaxSleep)); + noperations++; + } + globalTransmitter->DeRegister(r); + { + SGSharedPtr tn(new TestBaseNotification()); + SG_CHECK_EQUAL(globalTransmitter->Count(), 0); + globalTransmitter->NotifyAll(tn); + SG_CHECK_EQUAL(globalTransmitter->Count(), 0); + } + summary(timeStamp, globalTransmitter, "base"); +} + +void testEmesaryMultipleRecipients() +{ + printf("Testing multiple recipients\n"); + + simgear::Emesary::Transmitter* globalTransmitter = simgear::Emesary::GlobalTransmitter::instance(); + SGTimeStamp timeStamp; + timeStamp.stamp(); + + + + // create and register test recipients + std::vector recips; + for (int i = 0; i < 5; i++) + { + TestBaseRecipient *newRecipient = new TestBaseRecipient; + recips.push_back(newRecipient); + globalTransmitter->Register(newRecipient); + } + + // check that TestThreadNotification are ignored + int rcount = globalTransmitter->SentMessageCount(); + SG_CHECK_EQUAL(globalTransmitter->Count(), recips.size()); + { + // send a bunch of notifications. + for (int i = 0; i < MaxIterationsBaseMultipleRecipients; i++) + { + SGSharedPtr ttn(new TestThreadNotification("TestThread")); + globalTransmitter->NotifyAll(ttn); + //System.Threading.Thread.Sleep(rng.Next(MaxSleep)); + } + std::for_each(recips.begin(), recips.end(), [&globalTransmitter](TestBaseRecipient* r) { + SG_CHECK_EQUAL(0, r->receiveCount); + }); + } + SG_CHECK_NE(rcount, globalTransmitter->SentMessageCount()); + + rcount = globalTransmitter->SentMessageCount(); + { + SGSharedPtr < TestBaseNotification>tbn(new TestBaseNotification()); + for (int i = 0; i < MaxIterationsBaseMultipleRecipients; i++) + { + tbn->index = i; + globalTransmitter->NotifyAll(tbn); + noperations++; + } + std::for_each(recips.begin(), recips.end(), [&globalTransmitter](TestBaseRecipient* r) { + SG_CHECK_EQUAL(r->receiveCount, MaxIterationsBaseMultipleRecipients); + }); + + } + // now degregister all + std::for_each(recips.begin(), recips.end(), [&globalTransmitter](TestBaseRecipient* r) { + globalTransmitter->DeRegister(r); + }); + + SG_CHECK_EQUAL(globalTransmitter->Count(), 0); + { + SGSharedPtr < TestThreadNotification >ttn(new TestThreadNotification("TestThread")); + globalTransmitter->NotifyAll(ttn); + } + SG_CHECK_EQUAL(globalTransmitter->Count(), 0); + + summary(timeStamp, globalTransmitter, "base"); +} int main(int ac, char ** av) { + { + SGSharedPtr tn(new TestDerefNotification()); + } + SG_CHECK_EQUAL(TestDerefNotification_destructor_count, 1); + + // should still be deleted when the inner pointer goes out of scope. + auto tn1 = new TestDerefNotification(); + { + SGSharedPtr tn(tn1); + } + SG_CHECK_EQUAL(TestDerefNotification_destructor_count, 2); + + testEmesaryBase(); + + testEmesaryMultipleRecipients(); + testEmesaryThreaded(); + testEmesaryThreadedAddDuringReceive(); + std::cout << "all tests passed" << std::endl; return 0; } diff --git a/simgear/nasal/cppbind/NasalEmesaryInterface.hxx b/simgear/nasal/cppbind/NasalEmesaryInterface.hxx index d9411fa1..1af86abb 100644 --- a/simgear/nasal/cppbind/NasalEmesaryInterface.hxx +++ b/simgear/nasal/cppbind/NasalEmesaryInterface.hxx @@ -71,10 +71,9 @@ namespace nasal } std::atomic receiveCount; - virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotification &n) + virtual simgear::Emesary::ReceiptStatus Receive(simgear::Emesary::INotificationPtr n) { - - simgear::Notifications::MainLoopNotification *mln = dynamic_cast(&n); + auto mln = dynamic_pointer_cast(n); if (mln) { switch (mln->GetValue()) { @@ -101,8 +100,8 @@ namespace nasal } return simgear::Emesary::ReceiptStatus::OK; } + auto gccn = dynamic_pointer_cast(n); - auto *gccn = dynamic_cast(&n); if (gccn) { CanWait = gccn->GetCanWait(); Active = gccn->GetActive(); diff --git a/simgear/nasal/cppbind/detail/to_nasal_helper.cxx b/simgear/nasal/cppbind/detail/to_nasal_helper.cxx index 14ffe909..6c339c94 100644 --- a/simgear/nasal/cppbind/detail/to_nasal_helper.cxx +++ b/simgear/nasal/cppbind/detail/to_nasal_helper.cxx @@ -30,15 +30,7 @@ namespace nasal // create single instance of the main loop recipient for Nasal - this will self register at the // global transmitter - and that's all that is needed to link up the background GC to the main // loop in FG that will send out the MainLoop notifications. - //class NasalMainLoopRecipientSingleton : public simgear::Singleton - //{ - //public: - // NasalMainLoopRecipientSingleton() - // { - // } - // virtual ~NasalMainLoopRecipientSingleton() {} - //}; - SGSharedPtr mrl; + NasalMainLoopRecipient mrl; //---------------------------------------------------------------------------- naRef to_nasal_helper(naContext c, const std::string& str) diff --git a/simgear/structure/SGSharedPtr.hxx b/simgear/structure/SGSharedPtr.hxx index d9f5540d..00459f84 100644 --- a/simgear/structure/SGSharedPtr.hxx +++ b/simgear/structure/SGSharedPtr.hxx @@ -44,7 +44,7 @@ class SGWeakPtr; /// ordinary pointers or SGWeakPtr's. /// There is a very good description of OpenSceneGraphs ref_ptr which is /// pretty much the same than this one at -/// http://dburns.dhs.org/OSG/Articles/RefPointers/RefPointers.html +/// https://web.archive.org/web/20061012014801/http://dburns.dhs.org/OSG/Articles/RefPointers/RefPointers.html template class SGSharedPtr { @@ -158,6 +158,12 @@ SGSharedPtr static_pointer_cast(SGSharedPtr const & r) return SGSharedPtr( static_cast(r.get()) ); } +template +SGSharedPtr dynamic_pointer_cast(SGSharedPtr const& r) +{ + return SGSharedPtr(dynamic_cast(r.get())); +} + /** * Compare two SGSharedPtr objects for equality. *