Author: derek Date: Tue Sep 30 22:35:22 2008 New Revision: 440 Modified: sofaswitch/trunk/am/comms.cxx sofaswitch/trunk/am/comms.h sofaswitch/trunk/am/manager.cxx sofaswitch/trunk/am/manager.h sofaswitch/trunk/am/supervisor.cxx sofaswitch/trunk/mcu/comms.cxx sofaswitch/trunk/mcu/comms.h sofaswitch/trunk/mcu/manager.cxx sofaswitch/trunk/mcu/manager.h sofaswitch/trunk/sofa/include/sofa/messages.h sofaswitch/trunk/sofa/include/sofa/messlist.h sofaswitch/trunk/sofa/src/messages.cxx sofaswitch/trunk/sofa/src/messlist.cxx sofaswitch/trunk/voip/comms.cxx sofaswitch/trunk/voip/comms.h sofaswitch/trunk/voip/counts.cxx sofaswitch/trunk/voip/counts.h sofaswitch/trunk/voip/jocall.cxx sofaswitch/trunk/voip/main.cxx sofaswitch/trunk/voip/mymanager.cxx sofaswitch/trunk/voip/mymanager.h sofaswitch/trunk/voip/supervisor.cxx sofaswitch/trunk/voip/supervisor.h sofaswitch/trunk/voip/udpconnection.cxx sofaswitch/trunk/voip/udpconnection.h sofaswitch/trunk/voip/udpsound.cxx Log: lots and lots and lots of fixes, to bring it up date with the latest ptlib/opal combination. Improve reliability - add read/write locks. Modified: sofaswitch/trunk/am/comms.cxx ============================================================================== --- sofaswitch/trunk/am/comms.cxx (original) +++ sofaswitch/trunk/am/comms.cxx Tue Sep 30 22:35:22 2008 @@ -69,7 +69,9 @@ PTRACE(3, "Comms::End of this Comms thread for " << GetOurId() << " Media streams closed"); - new CommsConsumer(manager, this); + MessageFrame::OnTermination(); + + new CommsConsumer(manager, GetOurId()); } void Comms::ProcessTransferUdp(const PStringArray &cmds, const PString & sender) @@ -265,7 +267,7 @@ : MessageFrameListProvider(_lid), manager(_manager) { - StartWorkNow(); + StartWorkerNow(); } MessageFrame *CommsProvider::BuildMessageFrame() @@ -281,20 +283,20 @@ Comms *CommsProvider::GetNextComms() { - for(;;) { - Comms *c = (Comms *)GetTheNextMessageFrame(); - if (c->IsTerminated()) { - manager.ThisFrameHasDied(c); + Comms *c = NULL; + for(;;) { + c = (Comms *)GetTheNextMessageFrame(); + if (!c->IsTerminated()) + return c; } return c; - } } ///////////////////////////////////////////////////////////////////////////// -CommsConsumer::CommsConsumer(Manager &_manager, Comms *m) +CommsConsumer::CommsConsumer(Manager &_manager, PString _idToKill) :PThread(10000, AutoDeleteThread, LowPriority, "%X Comms delete"), manager(_manager), - comms(m) + idToKill(_idToKill) { IncCountCommsConsumers(); Resume(); @@ -305,23 +307,19 @@ DecCountCommsConsumers(); } +/*We Know the comms instance has died - the thread has terminated. Since the comms thread has terminated, we are running, so we have to get the comms instance from the manager's lists and delete it.. */ void CommsConsumer::Main() { - if (comms == NULL) - return; - - comms->WaitForTermination(); - PString ourId = comms->GetOurId(); - PTRACE(3, "Silently end all Removed comms id " << ourId); + PTRACE(3, "Silently end all Removed comms id " << idToKill); PINDEX iteration = 0; for(;;) { iteration++; - PTRACE(6, "look for comms instance " << ourId << " in the manager" + PTRACE(6, "look for comms instance " << idToKill << " in the manager" << " iteration " << iteration); - Comms *c = manager.ClearCommsId(ourId); + Comms *c = manager.ClearCommsId(idToKill); if (c != NULL) { - PTRACE(3, "NOW delete comms instance " << ourId); + PTRACE(3, "NOW delete comms instance " << idToKill); PTRACE(3, "as we found it and have exclusive access to it"); PTRACE(3, "and know that this thread has finished."); delete c; Modified: sofaswitch/trunk/am/comms.h ============================================================================== --- sofaswitch/trunk/am/comms.h (original) +++ sofaswitch/trunk/am/comms.h Tue Sep 30 22:35:22 2008 @@ -249,8 +249,9 @@ PCLASSINFO(CommsConsumer, PThread); /**Constructer - create this class, which is given the Comms - instance to destroy. This consumer class then does the rest */ - CommsConsumer(Manager &_manager, Comms *destroyMe); + instance to destroy. This consumer class then does the rest + We have to get the id from the managers lists and then delete it.*/ + CommsConsumer(Manager &_manager, PString _idToKill); /**Destructor */ ~CommsConsumer(); @@ -262,8 +263,8 @@ /**A reference to the class that knows everything about everything*/ Manager & manager; - /**The instance we have to delete */ - Comms *comms; + /**The ID of the instance we have to delete */ + PString idToKill; }; Modified: sofaswitch/trunk/am/manager.cxx ============================================================================== --- sofaswitch/trunk/am/manager.cxx (original) +++ sofaswitch/trunk/am/manager.cxx Tue Sep 30 22:35:22 2008 @@ -54,14 +54,11 @@ ourId(_ourId), runningComms(NULL) { - activeCalls.Initialise(); - bogusIdCount.SetValue(0); runningComms = new CommsProvider(lid, *this); runningComms->SetPreAllocatedSize(10); - runningComms->Initialise(); - deadCommsInstances.Initialise(); + nodesMakingCalls.StartWorkerNow(); } @@ -70,6 +67,10 @@ /* eradicate all current calls */ supervisor.Terminate(); supervisor.WaitForTermination(); + nodesMakingCalls.StopWorkerNow(); + runningComms->StopWorkerNow(); + + PThread::Sleep(2000); delete runningComms; runningComms = NULL; @@ -97,13 +98,13 @@ return c; } - c = (Comms *)deadCommsInstances.GetMessageFrameOn(memberId); + c = (Comms *)activeCalls.GetMessageFrameOn(memberId); if (c != NULL) { PTRACE(3, "Clear comms Id on " << memberId - << " for dead comms instances frame"); + << ");prebuilt, ready to go frames"); return c; } - + return NULL; } @@ -126,11 +127,12 @@ Comms * Manager::RemoveFromPreCallQueue(const PString & preferredId) { - Comms *c = NULL; - if (!preferredId.IsEmpty()) - c = (Comms *)nodesMakingCalls.GetMessageFrameOn(preferredId); - - return c; + PTRACE(3, "Manager\tRemove from preCall queue the ide " << preferredId); + Comms *c = NULL; + if (!preferredId.IsEmpty()) + c = (Comms *)nodesMakingCalls.GetMessageFrameOn(preferredId); + + return c; } Comms * Manager::CreateCall(const PString & remoteCont, const PString & preferredId) @@ -157,8 +159,7 @@ { /* Stop the building of new Comms instances */ PTRACE(3, "Manager\tRemove All Calls Start"); - runningComms->Terminate(); - runningComms->WaitForTermination(); + runningComms->StopWorkerNow(); runningComms->TerminateAllMessageFrames(); PTRACE(3, "Manager\tRemove all calls who are making calls"); @@ -186,8 +187,9 @@ { PStringStream result; - PStringArray currentNames = activeCalls.GetCurrentEntries(); PSortedStringList sortedNames; + PStringArray currentNames; + activeCalls.GetCurrentEntries(currentNames); for (PINDEX i = 0; i < currentNames.GetSize(); i++) sortedNames.AppendString(currentNames[i]); Modified: sofaswitch/trunk/am/manager.h ============================================================================== --- sofaswitch/trunk/am/manager.h (original) +++ sofaswitch/trunk/am/manager.h Tue Sep 30 22:35:22 2008 @@ -138,17 +138,10 @@ /**Report the current ids of active calls */ PString ActiveCallList(); - /**Count of the number of active calls in the system */ - PINDEX DeadCalls() { return deadCommsInstances.GetSize(); } - /**Go through the list of active calls, and tell them to "die now" */ void RemoveAllCalls(); - /**Place this Comms instance on the dead frames list */ - void ThisFrameHasDied(MessageFrame *c) - { deadCommsInstances.AddNewMessageFrame(c); } - protected: /**The interface to the external script which runs python, to provide comms for us */ @@ -166,10 +159,6 @@ entity */ MessageFrameList activeCalls; - /**A list of Comms instances that have ended. We just throw them onto - this list, and let CommsConsumer delete them as required */ - MessageFrameList deadCommsInstances; - /**An enity which could be used when logging in to the server */ PString ourPassword; Modified: sofaswitch/trunk/am/supervisor.cxx ============================================================================== --- sofaswitch/trunk/am/supervisor.cxx (original) +++ sofaswitch/trunk/am/supervisor.cxx Tue Sep 30 22:35:22 2008 @@ -90,10 +90,9 @@ << GetCountComms() << endl << PString("Consumer instance count ") << GetCountCommsConsumers() << endl - << PString("pre, active & dead calls ") + << PString("pre, active calls ") << ::setw(6) << PString(mgr->PendingCalls()) << ::setw(6) << PString(mgr->ActiveCalls()) - << ::setw(6) << PString(mgr->DeadCalls()) << endl << mgr->ActiveCallList(); Modified: sofaswitch/trunk/mcu/comms.cxx ============================================================================== --- sofaswitch/trunk/mcu/comms.cxx (original) +++ sofaswitch/trunk/mcu/comms.cxx Tue Sep 30 22:35:22 2008 @@ -63,7 +63,10 @@ void Comms::OnTermination() { - new CommsConsumer(manager, this); + PSafePtr <Member> member = manager.activeMembers.FindWithLock(GetOurId(), PSafeReadOnly); + if (member != NULL) + member->Clear(); + new CommsConsumer(manager, GetOurId()); } void Comms::ProcessTransferUdp(const PStringArray &cmds, const PString & sender) @@ -222,7 +225,7 @@ : MessageFrameListProvider(_lid), manager(_manager) { - StartWorkNow(); + StartWorkerNow(); } MessageFrame *CommsProvider::BuildMessageFrame() @@ -237,21 +240,22 @@ Comms *CommsProvider::GetNextComms() { - for(;;) { - Comms *c = (Comms *)GetTheNextMessageFrame(); - if (c->IsTerminated()) { - manager.ThisFrameHasDied(c); - } else - return c; - } + Comms *c = NULL; + for(;;) { + c = (Comms *)GetTheNextMessageFrame(); + if (!c->IsTerminated()) { + return c; + } + } + return NULL; } ///////////////////////////////////////////////////////////////////////////// -CommsConsumer::CommsConsumer(Manager & _manager, Comms *m) - :PThread(10000, AutoDeleteThread, LowPriority, "%X Comms delete"), - manager(_manager), - comms(m) +CommsConsumer::CommsConsumer(Manager & _manager, PString & _idToKill) +:PThread(10000, AutoDeleteThread, LowPriority, "%X Comms delete"), + manager(_manager), + idToKill(_idToKill) { manager.CommsConsumerRunning(); Resume(); @@ -264,34 +268,29 @@ void CommsConsumer::Main() { - if (comms == NULL) - return; - - comms->WaitForTermination(); - PString ourId = comms->GetOurId(); - PTRACE(3, "Silently end all Removed comms id"); - - PINDEX iteration = 0; + PTRACE(3, "Silently end all Removed comms id" << idToKill); + + PINDEX iteration = 0; for(;;) { - iteration++; - PTRACE(6, "look for comms instance " << ourId << " in the manager" + iteration++; + PTRACE(6, "look for comms instance " << idToKill << " in the manager" << " iteration " << iteration); - Comms *c = manager.ClearCommsId(ourId); - if (c != NULL) { - PTRACE(3, "NOW delete comms instance " << ourId); - PTRACE(3, "as we found it and have exclusive access to it"); - PTRACE(3, "and know that this thread has finished."); - delete c; - break; - } - PThread::Sleep(400); + Comms *c = manager.ClearCommsId(idToKill); + if (c != NULL) { + PTRACE(3, "NOW delete comms instance " << idToKill); + PTRACE(3, "as we found it and have exclusive access to it"); + PTRACE(3, "and know that this thread has finished."); + delete c; + break; + } + PThread::Sleep(400); } - - PTRACE(3, "Remove member " << ourId << " from the conference room"); - manager.RemoveMemberId(ourId); - PTRACE(3, "Removed " << ourId << " from activeMembers in the manager"); -} + PTRACE(3, "Remove member " << idToKill << " from the conference room"); + manager.RemoveMemberId(idToKill); + PTRACE(3, "Removed " << idToKill << " from activeMembers in the manager"); +} + /* The comment below is magic for those who use emacs to edit this file. */ /* With the comment below, the tab key does auto indent to 2 spaces. */ Modified: sofaswitch/trunk/mcu/comms.h ============================================================================== --- sofaswitch/trunk/mcu/comms.h (original) +++ sofaswitch/trunk/mcu/comms.h Tue Sep 30 22:35:22 2008 @@ -211,7 +211,7 @@ /**Constructer - create this class, which is given the Comms instance to destroy. This consumer class then does the rest */ - CommsConsumer(Manager & _manager, Comms *destroyMe); + CommsConsumer(Manager & _manager, PString & _idToKill); /**Destructor */ ~CommsConsumer(); @@ -223,8 +223,8 @@ /**The manager instance, that knows everything */ Manager & manager; - /**The instance we have to delete */ - Comms *comms; + /**The string Id of the instance we have to delete */ + PString idToKill; }; #endif // _Mcu_COMMS_H Modified: sofaswitch/trunk/mcu/manager.cxx ============================================================================== --- sofaswitch/trunk/mcu/manager.cxx (original) +++ sofaswitch/trunk/mcu/manager.cxx Tue Sep 30 22:35:22 2008 @@ -66,12 +66,9 @@ { bogusIdCount.SetValue(0); runningComms = new CommsProvider(lid, *this); - runningComms->Initialise(); runningComms->SetPreAllocatedSize(10); - nodesMakingCalls.Initialise(); - activeCalls.Initialise(); - deadCommsInstances.Initialise(); + nodesMakingCalls.StartWorkerNow(); PThread::Create(PCREATE_NOTIFIER(GarbageMain), 0, PThread::AutoDeleteThread, @@ -85,18 +82,22 @@ PTRACE(3, "Manager destructor"); supervisor.Terminate(); supervisor.WaitForTermination(); - - delete runningComms; - runningComms = NULL; + nodesMakingCalls.StopWorkerNow(); + runningComms->StopWorkerNow(); garbageExit.Signal(); garbageFinished.Wait(); + + PThread::Sleep(2000); + delete runningComms; + runningComms = NULL; } Comms * Manager::ClearCommsId(const PString &memberId) { PTRACE(3, "Clear comms Id on " << memberId); + Comms *c = (Comms *)activeCalls.GetMessageFrameOn(memberId); if (c != NULL) { PTRACE(3, "Clear comms Id on " << memberId << " for active calls"); @@ -115,14 +116,7 @@ << " for prebuilt, ready to go frames"); return c; } - - c = (Comms *)deadCommsInstances.GetMessageFrameOn(memberId); - if (c != NULL) { - PTRACE(3, "Clear comms Id on " << memberId - << " for dead comms instances frame"); - return c; - } - + return NULL; } @@ -309,8 +303,7 @@ void Manager::RemoveAllCalls() { PTRACE(3, "Remove All callls from the manager, close running comms down"); - runningComms->Terminate(); - runningComms->WaitForTermination(); + runningComms->StopWorkerNow(); nodesMakingCalls.TerminateAllMessageFrames(); activeCalls.TerminateAllMessageFrames(); Modified: sofaswitch/trunk/mcu/manager.h ============================================================================== --- sofaswitch/trunk/mcu/manager.h (original) +++ sofaswitch/trunk/mcu/manager.h Tue Sep 30 22:35:22 2008 @@ -145,10 +145,6 @@ instances running */ void CommsConsumerStopped() { --commsConsumerCount; } - /**Place this Comms instance on the dead frames list */ - void ThisFrameHasDied(MessageFrame *c) - { deadCommsInstances.AddNewMessageFrame(c); } - /**Callback for removing a connection from the list of active connections */ void OnReleased(const PString & memberId); @@ -189,11 +185,6 @@ entity */ MessageFrameList activeCalls; - /**A list of Comms instances that have ended. We just throw them onto - this list, and let CommsConsumer delete them as required */ - MessageFrameList deadCommsInstances; - - /**The unique ID that identifies the remote controller who is responsible for the overview of this sofaswitch */ PString controllerId; @@ -211,7 +202,7 @@ /**The list of initialised, ready to go, Comms instances */ CommsProvider *runningComms; - /**@name Clean up dead activeMembers */ + /**@name Clean up pre and activeMembers */ //@{ /**The syncpoint that flags when we exit, and is used to time the 1 second gaps between cleaning */ Modified: sofaswitch/trunk/sofa/include/sofa/messages.h ============================================================================== --- sofaswitch/trunk/sofa/include/sofa/messages.h (original) +++ sofaswitch/trunk/sofa/include/sofa/messages.h Tue Sep 30 22:35:22 2008 @@ -98,7 +98,7 @@ PString GetLastSender() { return lastSender; } /**Report the id used by this connection */ - PString GetOurId() { return ourId; } + PString & GetOurId() { return ourId; } /**Close this thread that is doing the Processing of incoming Messages.*/ virtual void Terminate(); Modified: sofaswitch/trunk/sofa/include/sofa/messlist.h ============================================================================== --- sofaswitch/trunk/sofa/include/sofa/messlist.h (original) +++ sofaswitch/trunk/sofa/include/sofa/messlist.h Tue Sep 30 22:35:22 2008 @@ -77,12 +77,25 @@ }; ///////////////////////////////////////////////////////////////////////////// +/**Create the list that we use to whold the messageframe pointers, + indexted by PString elements */ +typedef std::map<PString, MessageFrame *> MFL; +////////////////////////////////////////////////////////////////////////////// -PDECLARE_LIST (MessageFrameList, MessageFrame *) -#ifdef DOC_PLUS_PLUS //This makes emacs bracket matching code happy. -/** A list of all MessageFrame classes, which have been created and are waiting - for use. After use in one call, they are deleted. +/** Extract out the contents of the pdeclare_dicionary macro as + something meaningfull for doxygen to parse + + A list of all MessageFrame classes, which have been created and + are waiting for use. + + This class is always a descendant of MessageFrameList, and in so + doing we can have a constructor that we can access. + + Add all the worker methods requried for handling a list of + MessageFrame pointers + + MessageFrame instances are used once, and are never reused in a second place. This class provides a second purpose, to store MessageFrame structure that have been allocated and used in a previous call. In this @@ -90,27 +103,32 @@ rather than inside the MyOpalCall destructor, so that there is no possibility of holding up the MyOpalCall destructor - Access to this list is kept threadsafe by use of a mutex*/ -class MessageFrameList : public MessageFrame * + Access to this list is kept threadsafe by use of a mutex +*/ +class MessageFrameList : public PObject { -#endif + PCLASSINFO(MessageFrameList, PObject); + public: + /**Constructor, so that entries are not deleted on removal */ + + MessageFrameList(); /**Destroy all elements in this list */ ~MessageFrameList(); - - /**Add an instance of the MessageFrame class (supplied as an argument) to the end of - this list*/ + + /**Add an instance of the MessageFrame class (supplied as an + argument) to the end of this list*/ virtual void AddNewMessageFrame(MessageFrame *src); - - /**True if this frame list is empty*/ - bool Empty() { return GetSize() == 0; } - + /**Return the first available MessageFrame structure, which will have any ID. Repeated calls to this will empty this list. @return NULL is returned if the list is empty*/ virtual MessageFrame *GetTheNextMessageFrame(); + /**True if this frame list is empty*/ + bool Empty() { return GetSize() == 0; } + /**Return the next available MessageFrame structure, which will have the specified ID*/ MessageFrame * GetMessageFrameOn(const PString & id); @@ -122,153 +140,161 @@ void TerminateAllMessageFrames(); /**Thread safe read of the number of elements on this list. */ - virtual PINDEX GetSize() { PWaitAndSignal m(mutex); return PAbstractList::GetSize(); } + virtual PINDEX GetSize(); /**Report the names of the nodes sitting in this list */ - PStringArray GetCurrentEntries(); - - /**Initialise, so it will not delete frames on removing them from the list */ - void Initialise(); + void GetCurrentEntries(PStringArray & answer); protected: + /**The list of frames in this class. This list is a STL thing - + fastest list in C++ we can have. The list is indexed by PString + elements. */ + MFL frameList; + /**Local variable which protects access. */ PMutex mutex; }; - //////////////////////////////////////////////////////////////////////////////// -/**A class that extends MessageFrameList by adding a thread to create, - ahead of time, MessageFrame instances, that are initialised with - the remote server and are all ready to go */ -class MessageFrameListProvider: public MessageFrameList +/**A class that extends MessageFrameList by adding a thread to do some + task, at a regular rate. A descendant class will add the action + method + + Note that the action thread and descendants will delete the + MessageFrame if it has gone bad.. Thus, if this list holds things + other than MessageFrames, they have to cope with being deleted + "wherever"*/ +class MessageFrameListAction: public MessageFrameList { - PCLASSINFO(MessageFrameListProvider, MessageFrameList); - - /**Constructor - does nothing. The work of initialisation etc is left to the Initialise method */ - MessageFrameListProvider(LaunchNewId &_lid); - - /**Destructor. Destroy all pending instances */ - ~MessageFrameListProvider(); - - /**End this thread now */ - virtual void Terminate(); - - /**Return when this thread has finished */ - virtual void WaitForTermination(); - - /**Set the size of this list, so there is a sufficient number of - preallocated frames. The default value is 10. */ - void SetPreAllocatedSize(PINDEX newSize = 10); - - /**get the next message frame on this list. When returned, we know - it is running, and initialized, and ready to go. However, it - could receive a "hangup" signal at any time from the remote - controller. So, we have no way of knowing if this class is still - running at the time we return it. Consequently, there are no - guarantees this message frame is running */ - MessageFrame * GetTheNextMessageFrame(); - - protected: - - /**Do the work of actually building a new MessageFrame. This method - will be overridden, to create descendants of MessageFrames, and - store them in this list. - - We DEMAND that the user provides their own interpretation of this - method. */ - virtual MessageFrame *BuildMessageFrame() = 0; + PCLASSINFO(MessageFrameListAction, MessageFrameList); + public: + /**Constructor - does nothing. It must not start work, as a + descendant of this class will have problems (virtual method + called and other warnings, cause the thread is running before the + descendant is actually finished being built.*/ + MessageFrameListAction(); + + /**Destructor. Terminate, but not destroy, all pending instances */ + ~MessageFrameListAction(); + + /**Cause each entry on this list to Terminate now. The thread in + each message frame entry works out how to delete itself. */ + void RemoveAllMessageFrames(); /**the work of building a new message frame is started by the descendant class calling this method. Thus, the constructor of the descendant class should call this method. The constructor of this MessageFrameListProvider MUST not call this method. You will get pure virtual errors, as BuildMessageFrame is not defined. */ - void StartWorkNow(); + void StartWorkerNow(); + + /**End the worker thread which is running in this class. This is the + opposite of StartWorkerNow() */ + void StopWorkerNow(); + + /**Set the interval (in ms) between process of this list by the + internal worker thread */ + void SetActionInterval(PINDEX newInterval); + + /**Go through the list of frames. Any that have terminated are dead, and so + delete them */ + void RemoveDeadEntries(); + + + protected: #ifdef DOC_PLUS_PLUS /**The thread, which runs to create new and ready to go instances, runs in here */ - void MakeNewMessageFrames(PThread &, INT param); + void ActionMain(PThread &, INT param); #else - PDECLARE_NOTIFIER(PThread, MessageFrameListProvider, MakeNewMessageFrames); + PDECLARE_NOTIFIER(PThread, MessageFrameListAction, ActionMain); #endif - /**Go through the list of frames. Any that have terminated are dead, and so - delete them */ - void RemoveDeadEntries(); + void MakeNewMessageFrames(); + + /**Do the work required for this class in this method, which all descendant + classes must override */ + virtual void ActionThreadHere() { } /**The thread which will create new instances of MessageFrame structures, as required */ - PThread *makeNewMessageFrame; - - /**The flag which we use to indicate to the makeNewMessageFrame - thread that it is time to build a new frame */ - PSyncPoint makeAnother; + PThread *actionThread; /**A flag to indicate it is time to end this thread */ bool keepGoing; - /**Reference to the LaunchNewId class, which is required for building - Comms classes */ - LaunchNewId & lid; + /**The time (in ms) that this list is processed by the worker thread */ + PINDEX actionPeriod; + + /**The sync point which drives the worker thread to wake up and do something */ + PSyncPoint runAgain; +}; + + +///////////////////////////////////////////////////////////////////////////// +/**Build frames and add them */ +class MessageFrameListProvider : public MessageFrameListAction +{ + PCLASSINFO(MessageFrameListProvider, MessageFrameListAction); + public: + /**Constructor */ + MessageFrameListProvider(LaunchNewId &_lid); + + /**Destructro */ + ~MessageFrameListProvider(); + + /**Set the size of this list, so there is a sufficient number of + preallocated frames. The default value is 10. */ + void SetPreAllocatedSize(PINDEX newSize = 10); + protected: + /**Do the work required in this method for this class, and build + MessageFrame instances */ + virtual void ActionThreadHere(); + + /**Do the work of actually building a new MessageFrame. This method + will be overridden, to create descendants of MessageFrames, and + store them in this list. + + We DEMAND that the user provides their own interpretation of this + method. */ + virtual MessageFrame *BuildMessageFrame() = 0; + + /**Reference to the LaunchNewId class, which is required for building + Comms classes */ + LaunchNewId & lid; /**The number of frames this list should hold, ready for use */ PINDEX requiredSize; -}; +}; ///////////////////////////////////////////////////////////////////////////// -/**A class that extends MessageFrameList by adding a thread to destroy old message frame instances. - This list contains those comms instances requested by the remote controller, which have yet to be - made live by connecting them with some remote node. */ -class MessageFrameListPending: public MessageFrameList +/**A class that extends MessageFrameList by adding a thread to terminate + old message frame instances. This list contains those comms + instances requested by the remote controller, which have yet to be + made live by connecting them with some remote node. + + This class cannot delete a messageframe instance. Message frames + are a separate thread, and will manage their own deletion.*/ +class MessageFrameListPending: public MessageFrameListAction { - PCLASSINFO(MessageFrameListPending, MessageFrameList); - - /**Constructor - Creates the garbage consumer thread, which peruses this list on a regular basis */ + PCLASSINFO(MessageFrameListPending, MessageFrameListAction); + + public: + /**Constructro */ MessageFrameListPending(); - /**Destructor. Destroy all instances in this list */ + /**Destructro */ ~MessageFrameListPending(); - - /**Add an instance of the MessageFrame class (supplied as an + + /**Add an instance of the MessageFrame class (supplied as an argument) to the end of this list. The date on the frame is set to NOW, and this frame has some number of seconds before it must be removed. The default period is 10 seconds.*/ virtual void AddNewMessageFrame(MessageFrame *src); - /**End this thread now */ - virtual void Terminate(); - - /**Return when this thread has finished */ - virtual void WaitForTermination(); - - /**Set the interval (in ms) between purges of this list by the - garbage collector */ - void SetPurgeInterval(PINDEX newInterval); - - protected: -#ifdef DOC_PLUS_PLUS -/**The thread, which runs to destroy old and unwanted MessageFrame instances */ - void RemoveOldMessageFrameMain(PThread &, INT param); -#else - PDECLARE_NOTIFIER(PThread, MessageFrameListPending, - RemoveOldMessageFrameMain); -#endif - - /**Do the work of the purger of old message frames */ - void PurgeOldMessageFrameBody(PINDEX maxTimeInterval); - - /**The thread which will remove old instances of MessageFrame structures, as required */ - PThread *removeOldMessageFrame; - - /**The flag which we use to indicate to the removeOldMessageFrame - thread that it is time to check for old unwanted frames */ - PSyncPoint removeAnother; - - /**A flag to indicate it is time to end this thread */ - bool keepGoing; - - /**The time (in ms) that this list is checked for old/dead frames */ - PINDEX delayPurgePeriod; + /**Do the work of the pending thread here - look for all the expired + MessageFrame instance and delete them */ + virtual void ActionThreadHere(); }; ///////////////////////////////////////////////////////////////////////////// Modified: sofaswitch/trunk/sofa/src/messages.cxx ============================================================================== --- sofaswitch/trunk/sofa/src/messages.cxx (original) +++ sofaswitch/trunk/sofa/src/messages.cxx Tue Sep 30 22:35:22 2008 @@ -137,7 +137,7 @@ { Terminate(); WaitForTermination(); - PTRACE(5, "DEstructor for MessageFrame " << ourId); + PTRACE(5, "messageframe\tDestructor for MessageFrame " << ourId); } void MessageFrame::DtmfReceivedFromTheWorld(const PString & receivedDtmf) @@ -178,6 +178,7 @@ void MessageFrame::Terminate() { + PTRACE(4, "MessageFrame " << GetOurId() << " is terminating now"); keepGoing = false; audioSocket.Close(); messagesSocket.Close(); Modified: sofaswitch/trunk/sofa/src/messlist.cxx ============================================================================== --- sofaswitch/trunk/sofa/src/messlist.cxx (original) +++ sofaswitch/trunk/sofa/src/messlist.cxx Tue Sep 30 22:35:22 2008 @@ -33,7 +33,9 @@ #include <sofa/messlist.h> - +MessageFrameList::MessageFrameList() +{ +} @@ -42,58 +44,40 @@ } -void MessageFrameList::Initialise() -{ - DisallowDeleteObjects(); -} - - -PStringArray MessageFrameList::GetCurrentEntries() +void MessageFrameList::GetCurrentEntries(PStringArray & result) { - PSortedStringList result; PWaitAndSignal m(mutex); - - for (PINDEX i = 0; i < PAbstractList::GetSize(); i++) { - PString nodeName = ((MessageFrame *)PAbstractList::GetAt(i))->GetOurId(); - result.AppendString(nodeName); - } - - return result; + MFL::iterator it; + for (it = frameList.begin(); it != frameList.end(); ++it) + result.AppendString(it->second->GetOurId()); } void MessageFrameList::TerminateMessageFrame(const PString & id) { PWaitAndSignal m(mutex); - if (PAbstractList::GetSize() == 0) { - return ; - } - - for (PINDEX i = 0; i < PAbstractList::GetSize(); i++) - if (((MessageFrame *)PAbstractList::GetAt(i))->GetOurId() == id) { - PObject *p = PAbstractList::GetAt(i); - MessageFrame * c = (MessageFrame *)p; - PTRACE(3, "Advise the requested comms object on " << id << " to terminate"); - c->Terminate(); - return; - } + if(frameList.size() == 0) + return; - return; + MFL::iterator it = frameList.find(id); + if (it != frameList.end()) { + PTRACE(3, "Advise the requested comms object on " << id + << " to terminate"); + it->second->Terminate(); + } } void MessageFrameList::TerminateAllMessageFrames() { PWaitAndSignal m(mutex); - if (PAbstractList::GetSize() == 0) { - return ; - } - - for (PINDEX i = PAbstractList::GetSize(); i > 0; i--) { - PObject *p = PAbstractList::GetAt(i - 1); - MessageFrame * c = (MessageFrame *)p; - PTRACE(3, "Advise the requested comms object on " << c->GetOurId() << " to terminate"); - c->Terminate(); + MFL::const_iterator it; + + for (it = frameList.begin(); it != frameList.end(); it++) { + PTRACE(3, "Advise the requested comms object on " + << it->second->GetOurId() << " to terminate"); + it->second->Terminate(); + } return; @@ -103,44 +87,46 @@ { PWaitAndSignal m(mutex); - if (PAbstractList::GetSize() == 0) { + MFL::iterator it = frameList.find(id); + if (it == frameList.end()) return NULL; - } - - for (PINDEX i = 0; i < PAbstractList::GetSize(); i++) - if (((MessageFrame *)PAbstractList::GetAt(i))->GetOurId() == id) { - PObject *p = PAbstractList::RemoveAt(i); - MessageFrame * c = (MessageFrame *)p; - PTRACE(3, "Provide the requested comms object on " << id << " as requested"); - return c; - } - - return NULL; + + MessageFrame * c = it->second; + + frameList.erase(it); + + return c; } +PINDEX MessageFrameList::GetSize() +{ + PWaitAndSignal m(mutex); + return frameList.size(); +} MessageFrame *MessageFrameList::GetTheNextMessageFrame() { PWaitAndSignal m(mutex); - if (PAbstractList::GetSize() == 0) { + if (frameList.size() == 0) { PTRACE(3, "No more available comms to use"); return NULL; } - for (PINDEX i = 0; i < PAbstractList::GetSize(); i++) { - PObject *p = PAbstractList::RemoveAt(i); - MessageFrame * c = (MessageFrame *)p; - PTRACE(3, "Provide the requested comms object on " << c->GetOurId() - << " as requested"); - return c; + MFL::iterator it; + for (it = frameList.begin(); it != frameList.end();) { + MessageFrame * c = (MessageFrame *)it->second; + if (!it->second->IsTerminated()) { + frameList.erase(it++); + return c; + } else + ++it; } return NULL; } - void MessageFrameList::AddNewMessageFrame(MessageFrame *newEntry) { if (newEntry == NULL) { @@ -149,245 +135,156 @@ } PWaitAndSignal m(mutex); - Append(newEntry); - PTRACE(3, " MessageFrame list has grown to a size of " << PAbstractList::GetSize()); + frameList.insert(make_pair(newEntry->GetOurId(), newEntry)); + PTRACE(3, " MessageFrame list has grown to a size of " << frameList.size()); } -//////////////////////////////////////////////////////////////////////////////// -MessageFrameListProvider::MessageFrameListProvider(LaunchNewId &_lid) - : lid(_lid) +///////////////////////////////////////////////////////////////////////////// +MessageFrameListAction::MessageFrameListAction() { - SetPreAllocatedSize(10); keepGoing = true; - MessageFrameList::Initialise(); +} - PTRACE(3, "Create new MessageFrame provider."); +MessageFrameListAction::~MessageFrameListAction() +{ + if (actionThread != NULL) { + StopWorkerNow(); + delete actionThread; + actionThread = NULL; + } } -void MessageFrameListProvider::StartWorkNow() + +void MessageFrameListAction::StartWorkerNow() { - makeNewMessageFrame = PThread::Create(PCREATE_NOTIFIER(MakeNewMessageFrames), + actionThread = PThread::Create(PCREATE_NOTIFIER(ActionMain), 0, PThread::NoAutoDeleteThread, PThread::NormalPriority, "%X make Frames"); } -void MessageFrameListProvider::RemoveDeadEntries() +void MessageFrameListAction::StopWorkerNow() { - MessageFrameList deads; - - { - PWaitAndSignal m(mutex); - for(PINDEX i = PAbstractList::GetSize(); i > 0; i--) { - PObject *p = PAbstractList::GetAt(i - 1); - MessageFrame * c = (MessageFrame *)p; - if (c == NULL) - continue; - if (c->IsTerminated()) { - deads.AddNewMessageFrame(c); - RemoveAt(i - 1); - } - } - } - while(deads.GetSize() > 0) { - PObject *f = deads.RemoveAt(0); - delete f; - } + keepGoing = false; + runAgain.Signal(); + actionThread->WaitForTermination(); } +void MessageFrameListAction::RemoveDeadEntries() +{ + PWaitAndSignal m(mutex); + MFL::iterator it; + for(it = frameList.begin(); it != frameList.end(); ) { + MessageFrame * c = it->second; + if (c->IsTerminated()) { + delete c; + frameList.erase(it++); + } else + ++it; + } +} -MessageFrameListProvider::~MessageFrameListProvider() +void MessageFrameListAction::ActionMain(PThread &, INT param) { - Terminate(); - makeNewMessageFrame->WaitForTermination(); - delete makeNewMessageFrame; - makeNewMessageFrame = NULL; + while (keepGoing) { + PTimeInterval waitPeriod(actionPeriod); - PThread::Sleep(1000); - PTRACE(3, "Finished destroy the message frame list provider"); + runAgain.Wait(waitPeriod); + ActionThreadHere(); + } } -void MessageFrameListProvider::SetPreAllocatedSize(PINDEX newSize) +void MessageFrameListAction::SetActionInterval(PINDEX newInterval) { - requiredSize = newSize; - makeAnother.Signal(); + runAgain.Signal(); + actionPeriod = newInterval; } - -void MessageFrameListProvider::MakeNewMessageFrames(PThread &, INT ) +//////////////////////////////////////////////////////////////////////////////// +MessageFrameListProvider::MessageFrameListProvider(LaunchNewId &_lid) + : lid(_lid) { - for( ;keepGoing; ) { - RemoveDeadEntries(); - while ((GetSize() < requiredSize) && keepGoing) { - PTRACE(3, "We have to build a new message frame "); - MessageFrame *mf = BuildMessageFrame(); - if (mf != NULL) { - PTRACE(3, "reqeustnewid Add message frame to list of " << *mf); - AddNewMessageFrame(mf); - } else { - PTRACE(3, "We had an error in buildmessageframe - null returned"); - } - } - for(PINDEX i = 0; (i < 10) && keepGoing; i++) - makeAnother.Wait(100); - } - PTRACE(3, "End of message frame list provider thread "); + SetActionInterval(1 * 1000); + SetPreAllocatedSize(10); + + PTRACE(3, "Create new MessageFrame provider."); } -void MessageFrameListProvider::Terminate() + + +MessageFrameListProvider::~MessageFrameListProvider() { - keepGoing = false; - makeAnother.Signal(); + StopWorkerNow(); + + PThread::Sleep(1000); + PTRACE(3, "Finished destroy the message frame list provider"); } -void MessageFrameListProvider::WaitForTermination() -{ - makeNewMessageFrame->WaitForTermination(); +void MessageFrameListProvider::SetPreAllocatedSize(PINDEX newSize) +{ + requiredSize = newSize; + runAgain.Signal(); } -MessageFrame * MessageFrameListProvider::GetTheNextMessageFrame() +void MessageFrameListProvider::ActionThreadHere() { - PTRACE(3, "Get new MessageFrame from prepared frame"); - MessageFrame *mf = NULL; - { - PWaitAndSignal m(mutex); - - makeAnother.Signal(); - while (keepGoing && (!Empty())) { - mf = (MessageFrame *)PAbstractList::RemoveAt(0); - if (mf == NULL) - continue; - - mf->SetCreatedAtNow(); - return mf; - } - } - - while (keepGoing) { - PTRACE(3, "The message frame list of prepared frames is empty"); - mf = BuildMessageFrame(); - if (mf == NULL) - continue; - - if (mf->IsTerminated()) { + PTRACE(5, "Start of MessageFrameListProvider::ActionThreadHere()"); + RemoveDeadEntries(); + while ((GetSize() < requiredSize) && keepGoing) { + PTRACE(3, "We have to build a new message frame "); + MessageFrame *mf = BuildMessageFrame(); + if (mf != NULL) { + PTRACE(3, "reqeustnewid Add message frame to list of " << *mf); AddNewMessageFrame(mf); - mf = NULL; - continue; + } else { + PTRACE(3, "We had an error in buildmessageframe - null returned"); } - - mf->SetCreatedAtNow(); - return mf; } - - PTRACE(3, "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX Return NULL"); - cerr << "return NULL here " << endl; - return NULL; } -/* - -MessageFrame *MessageFrameListProvider::BuildMessageFrame() -{ - PTRACE(3, "Build a mesasge frame "); - PTRACE(3, "MessageFrameListProvider::BuildMessageFrame()"); - MessageFrame *mf = new MessageFrame(lid, 0); - mf->Resume(); - mf->WaitUntilInitialised(); - return mf; -} -*/ - - + ///////////////////////////////////////////////////////////////////////////// MessageFrameListPending::MessageFrameListPending() { - MessageFrameList::Initialise(); - removeOldMessageFrame = PThread::Create(PCREATE_NOTIFIER(RemoveOldMessageFrameMain), 0, - PThread::NoAutoDeleteThread, - PThread::LowPriority, - PString("%X OldMessageFrame")); - - SetPurgeInterval(500); + SetActionInterval(500); } MessageFrameListPending::~MessageFrameListPending() { - Terminate(); - WaitForTermination(); - - PTRACE(5, "Delete old message frame thread pending thread"); - delete removeOldMessageFrame; - PTRACE(5, "old Message Frame Pending collector now deleted. "); -} - -void MessageFrameListPending::WaitForTermination() -{ - removeOldMessageFrame->WaitForTermination(); -} - -void MessageFrameListPending::Terminate() -{ - removeAnother.Signal(); -} - -void MessageFrameListPending::AddNewMessageFrame(MessageFrame *src) -{ - src->SetCreatedAtNow(); - MessageFrameList::AddNewMessageFrame(src); + StopWorkerNow(); } -void MessageFrameListPending::RemoveOldMessageFrameMain(PThread &, INT) +void MessageFrameListPending::ActionThreadHere() { - PTRACE(5, "old Message Frame Pending collector now START."); - while(!removeAnother.Wait(delayPurgePeriod)) - PurgeOldMessageFrameBody(10 * 1000); /* Pending frames can last for 10 seconds */ - - PurgeOldMessageFrameBody(0); - - PTRACE(5, "old Message Frame Pending collector now start to exit "); -} - -void MessageFrameListPending::PurgeOldMessageFrameBody(PINDEX maxTimeInterval) -{ - PTimeInterval maxTime(maxTimeInterval); - PWaitAndSignal m(mutex); - PTRACE(5, " One iteration of Message Frame garbage. Remove " << PAbstractList::GetSize()); - - - for (PINDEX i = PAbstractList::GetSize(); i > 0; i--) { - MessageFrame *possible = (MessageFrame *)PAbstractList::GetAt(i - 1); - PTRACE(5, possible->GetOurId() << " has been alive for " << possible->TimeSinceCreation()); - if (possible->TimeSinceCreation() > maxTime) { - PTRACE(5, " Message Frame pending delete " << possible->GetOurId()); + PTimeInterval maxTime(10 * 1000); + { + PWaitAndSignal m(mutex); + PTRACE(5, " One iteraiton of message pending garbage " << frameList.size()); + MFL::iterator it; + for (it = frameList.begin(); it != frameList.end(); ) { + MessageFrame *possible = it->second; + if ( (possible->TimeSinceCreation() > maxTime) || + (possible->IsTerminated() ) + ){ + PTRACE(5, " Message Frame pending terminate " << possible->GetOurId()); possible->Terminate(); - possible->WaitForTermination(); - PAbstractList::RemoveAt(i - 1); - delete possible; - continue; - } - PTRACE(5, "Message frame pending delete for " << possible->GetOurId() << " still alive"); - if (possible->IsTerminated()) { - PAbstractList::RemoveAt(i - 1); delete possible; - } else { - PTRACE(5, "Message frame pending delete for " << possible->GetOurId() << " skipped as still alive"); + frameList.erase(it++); + } + else + ++it; } } - - if (PAbstractList::GetSize() > 0) { - PTRACE(5, " One iteration of Message Frame pending. Remaining " - << PAbstractList::GetSize()); - } } - -void MessageFrameListPending::SetPurgeInterval(PINDEX newInterval) +void MessageFrameListPending::AddNewMessageFrame(MessageFrame *src) { - delayPurgePeriod = newInterval; + src->SetCreatedAtNow(); + MessageFrameList::AddNewMessageFrame(src); } Modified: sofaswitch/trunk/voip/comms.cxx ============================================================================== --- sofaswitch/trunk/voip/comms.cxx (original) +++ sofaswitch/trunk/voip/comms.cxx Tue Sep 30 22:35:22 2008 @@ -82,7 +82,7 @@ } void Comms::OnTermination() -{ +{ PTRACE(3, "END of comms communications for " << GetOurId()); } @@ -172,10 +172,12 @@ } } + PTRACE(3, sender << " " << cmds[0] << " " << cmds[1]); remoteController = sender; bool doAnswer = false; bool doAnswerDefined= false; if (cmds.GetSize() > 1) { + PTRACE(3, sender << " " << cmds[0] << " " << cmds[1]); if ((cmds[1] *= "true") || (cmds[1] *= "yes") || (cmds[1] *= "accept")) { doAnswer = true; doAnswerDefined = true; @@ -239,7 +241,7 @@ : MessageFrameListProvider(_lid), manager(_manager) { - StartWorkNow(); + StartWorkerNow(); } MessageFrame *CommsProvider::BuildMessageFrame() @@ -256,53 +258,46 @@ return mf; } -void CommsProvider::RemoveAllMessageFrames() -{ - PWaitAndSignal m(mutex); - Comms *mf = NULL; - - for(PINDEX i = 0; i < PAbstractList::GetSize(); i++) { - mf = (Comms *)PAbstractList::GetAt(i); - new CommsConsumer(mf); - } -} - Comms *CommsProvider::GetNextComms() { PTRACE(3, "GetNextComms from CommsProvider"); - for(;;) { - Comms *c = (Comms *)GetTheNextMessageFrame(); - if (c == NULL) - return NULL; - if (c->IsTerminated()) { - delete c; - } else { - return c; - } - } - return NULL; + + return (Comms *)GetTheNextMessageFrame(); } //////////////////////////////////////////////////////////////////////////////// -CommsConsumer::CommsConsumer(Comms *m) +CommsConsumer::CommsConsumer(MyManager & mgr, PString & _idToKill) :PThread(10000, AutoDeleteThread, NormalPriority, "%X Comms delete"), - comms(m) + manager(mgr), + idToKill(_idToKill) { - PTRACE(2, "Create a Clear Comms Consumer to clear a comms instance " << m->GetOurId()); + PTRACE(2, "Create a Clear Comms Consumer to clear a comms instance " + << idToKill); + IncCommsConsumer(); Resume(); } -void CommsConsumer::Main() +CommsConsumer::~CommsConsumer() { - if (comms == NULL) - return; - - comms->Terminate(); - - comms->WaitForTermination(); + DecCommsConsumer(); +} - delete comms; +void CommsConsumer::Main() +{ + PINDEX iteration = 0; + for(;;) { + iteration++; + Comms *c = manager.ClearCommsId(idToKill); + if (c != NULL) { + PTRACE(3, "NOW delete comms instance " << idToKill); + PTRACE(3, "as we found it and have exclusive access to it"); + PTRACE(3, "and know that this thread has finished."); + delete c; + return; + } + PThread::Sleep(400); + } } /* The comment below is magic for those who use emacs to edit this file. */ Modified: sofaswitch/trunk/voip/comms.h ============================================================================== --- sofaswitch/trunk/voip/comms.h (original) +++ sofaswitch/trunk/voip/comms.h Tue Sep 30 22:35:22 2008 @@ -188,10 +188,6 @@ /**Retrieve the next Comms instance */ Comms *GetNextComms(); - /**Cause each entry on this list to be deleted by an instance of the - CommsConsumer class */ - void RemoveAllMessageFrames(); - protected: /**Do the work of actually building a new MessageFrame */ @@ -213,15 +209,21 @@ /**Constructer - create this class, which is given the Comms instance to destroy. This consumer class then does the rest */ - CommsConsumer(Comms *destroyMe); + CommsConsumer(MyManager & mgr, PString & _idToKill); + /**destructor for statistics */ + ~CommsConsumer(); + /**Where this thread runs to do the work */ void Main(); protected: - /**The instance we have to delete */ - Comms *comms; + /** Link back to the master manager */ + MyManager & manager; + + /**The ID of the comms instance we have to delete */ + PString idToKill; }; ///////////////////////////////////////////////////////////////////////////// Modified: sofaswitch/trunk/voip/counts.cxx ============================================================================== --- sofaswitch/trunk/voip/counts.cxx (original) +++ sofaswitch/trunk/voip/counts.cxx Tue Sep 30 22:35:22 2008 @@ -37,6 +37,7 @@ ///////////////////////////////////////////////////////////////////////////// PAtomicInteger countComms; +PAtomicInteger countCommsConsumer; PAtomicInteger countEstablished; PAtomicInteger countMyManager; PAtomicInteger countMyOpalCall; @@ -46,6 +47,7 @@ PAtomicInteger countUdpSoundChannel; PAtomicInteger countDelComms; +PAtomicInteger countDelCommsConsumer; PAtomicInteger countDelEstablished; PAtomicInteger countDelMyManager; PAtomicInteger countDelMyOpalCall; @@ -62,6 +64,15 @@ << ::setw(6) << PString(countComms - countDelComms) << ::setw(0); return msg; } +PString GetCountCommsConsumer() +{ + PStringStream msg; + msg << ::setw(6) << PString(countCommsConsumer) + << ::setw(6) << PString(countDelCommsConsumer) + << ::setw(6) << PString(countCommsConsumer - countDelCommsConsumer) + << ::setw(0); + return msg; +} PString GetCountEstablished() { @@ -128,6 +139,7 @@ } void DecComms() { ++countDelComms; } +void DecCommsConsumer() { ++countDelCommsConsumer; } void DecEstablished() { ++countDelEstablished; } void DecMyManager() { ++countDelMyManager; } void DecMyOpalCall() { ++countDelMyOpalCall; } @@ -137,6 +149,7 @@ void DecUdpSoundChannel() { ++countDelUdpSoundChannel; } void IncComms() { ++countComms; } +void IncCommsConsumer() { ++countCommsConsumer; } void IncEstablished() { ++countEstablished; } void IncMyManager() { ++countMyManager; } void IncMyOpalCall() { ++countMyOpalCall; } Modified: sofaswitch/trunk/voip/counts.h ============================================================================== --- sofaswitch/trunk/voip/counts.h (original) +++ sofaswitch/trunk/voip/counts.h Tue Sep 30 22:35:22 2008 @@ -50,6 +50,16 @@ /**Report the count on the Audio Delay instances established */ PString GetCountAudioDelay(); +/**Decrement the count on the Comms Consumer instances established */ +void DecCommsConsumer(); + +/**Increment the count on the CommsConsumer instances established */ +void IncCommsConsumer(); + +/**Report the count on the CommsConsumer instances established */ +PString GetCountCommsConsumer(); + + /**Decrement the count on SofaSwitch Communication instances established */ void DecComms(); Modified: sofaswitch/trunk/voip/jocall.cxx ============================================================================== --- sofaswitch/trunk/voip/jocall.cxx (original) +++ sofaswitch/trunk/voip/jocall.cxx Tue Sep 30 22:35:22 2008 @@ -76,7 +76,8 @@ { PTRACE(3, "Destructor for Opal Call " << GetToken() << " and " << comms.GetOurId()); - new CommsConsumer(commsPtr); + manager.CallHasFinished(commsPtr); + new CommsConsumer(manager, comms.GetOurId()); PTRACE(3, "MyOpalCall\t" << *this << " destroyed dead and gone"); DecMyOpalCall(); } @@ -171,19 +172,23 @@ void MyOpalCall::AcceptThisCallMain(PThread &, INT) { - PTRACE(3, "Process request to accept this call " << *this); - for (PSafePtr<OpalConnection> conn(connectionsActive, PSafeReadOnly); conn != NULL; ++conn) { - PSafePtr<OpalUdpSsConnection> connection; - connection = PSafePtrCast<OpalConnection, OpalUdpSsConnection>(conn); - if (connection == NULL) { - if (acceptingThisCall) - conn->AnsweringCall(OpalConnection::AnswerCallNow); - else - conn->AnsweringCall(OpalConnection::AnswerCallDenied); - PTRACE(3, "Finished Answering the call for " << conn->GetToken()); + PTRACE(3, "Process request to accept this call " << *this); + for (PSafePtr<OpalConnection> conn(connectionsActive, PSafeReadOnly); conn != NULL; ++conn) { + PSafePtr<OpalUdpSsConnection> connection; + connection = PSafePtrCast<OpalConnection, OpalUdpSsConnection>(conn); + if (connection == NULL) { + if (acceptingThisCall) { + PTRACE(3, "Accept this call " << conn->GetToken()); + conn->AnsweringCall(OpalConnection::AnswerCallNow); + } else { + PTRACE(3, "Accept this call " << conn->GetToken()); + conn->AnsweringCall(OpalConnection::AnswerCallDenied); + } + PTRACE(3, "Finished Accepting (answering) the call for " + << conn->GetToken()); + } } - } - SafeDereference(); + SafeDereference(); } void MyOpalCall::CloseThisId() Modified: sofaswitch/trunk/voip/main.cxx ============================================================================== --- sofaswitch/trunk/voip/main.cxx (original) +++ sofaswitch/trunk/voip/main.cxx Tue Sep 30 22:35:22 2008 @@ -291,6 +291,7 @@ s->WaitUntilInitialised(); } while (s->IsTerminating()); + PTRACE(3, "SUPERVISOR is running ok now"); opal = new MyManager(lid, s,ourId, ourPassword); if (opal->Initialise(args)) { Modified: sofaswitch/trunk/voip/mymanager.cxx ============================================================================== --- sofaswitch/trunk/voip/mymanager.cxx (original) +++ sofaswitch/trunk/voip/mymanager.cxx Tue Sep 30 22:35:22 2008 @@ -86,10 +86,11 @@ PString & _ourId, PString & _ourPassword) : lid(_lid), supervisor(*_supervisor), - runningComms(*new CommsProvider(lid, *this)) + voipCommsProvider(*new CommsProvider(lid, *this)) { allocationNumber = 0; IncMyManager(); + nodesMakingCalls.StartWorkerNow(); ourId = _ourId; ourPassword = _ourPassword; @@ -107,19 +108,16 @@ #endif supervisor.Resume(); - nodesMakingCalls.Initialise(); + nodesMakingCalls.StartWorkerNow(); - runningComms.SetPreAllocatedSize(10); - runningComms.Initialise(); + voipCommsProvider.SetPreAllocatedSize(10); } MyManager::~MyManager() { - supervisor.Terminate(); - supervisor.WaitForTermination(); - delete & supervisor; - delete & runningComms; + nodesMakingCalls.StopWorkerNow(); + voipCommsProvider.StopWorkerNow(); PTRACE(3, "destroy the manager"); DecMyManager(); @@ -487,9 +485,12 @@ PString MyManager::GetStatusInfoSofaSwitchNode(const PString & token) { - PStringArray prebuilts = runningComms.GetCurrentEntries(); - PStringArray pendingNames = nodesMakingCalls.GetCurrentEntries(); - PSortedStringList currentNames = GetCurrentNames(); + PStringArray prebuilts; + voipCommsProvider.GetCurrentEntries(prebuilts); + PStringArray pendingNames; + nodesMakingCalls.GetCurrentEntries(pendingNames); + PSortedStringList currentNames; + GetCurrentNames(currentNames); PStringStream answer; if (prebuilts.GetStringsIndex(token) != P_MAX_INDEX) { @@ -528,7 +529,7 @@ PStringStream status; status << endl; status << "Pre built calls " - << PString(runningComms.GetSize()) << endl; + << PString(voipCommsProvider.GetSize()) << endl; status << "Pending message frames " << PString(nodesMakingCalls.GetSize()) << endl; status << "Current active calls " @@ -537,6 +538,8 @@ status << "--------------" << endl; status << " Comms count is " << GetCountComms() << endl; + status << " CommsConsumer count is " + << GetCountCommsConsumer() << endl; status << " Established count is " << GetCountEstablished() << endl; status << " MyManager count is " @@ -553,8 +556,10 @@ << GetCountUdpSoundChannel() << endl; status << "--------------" << endl; - PStringArray prebuilts = runningComms.GetCurrentEntries(); - PStringArray pendingNames = nodesMakingCalls.GetCurrentEntries(); + PStringArray prebuilts; + PStringArray pendingNames; + voipCommsProvider.GetCurrentEntries(prebuilts); + nodesMakingCalls.GetCurrentEntries(pendingNames); if (prebuilts.GetSize() > 0) { status << "PreBuilt frames " << endl; @@ -570,8 +575,8 @@ status << endl; } - PSortedStringList currentNames = GetCurrentNames(); - + PSortedStringList currentNames; + GetCurrentNames(currentNames); if (currentNames.GetSize() > 0) { status << "Current active calls " << endl; @@ -583,10 +588,8 @@ return status; } -PSortedStringList MyManager::GetCurrentNames() +void MyManager::GetCurrentNames(PSortedStringList & answer) { - PSortedStringList answer; - for (PSafePtr<OpalCall>call = activeCalls; call != NULL; ++call) { MyOpalCall *c = PSafePtrCast<OpalCall, MyOpalCall>(call); if (c != NULL) @@ -594,9 +597,12 @@ else answer.AppendString("No name - weird"); } - return answer; } +void MyManager::CallHasFinished(Comms *c) +{ + finishedCalls.AddNewMessageFrame(c); +} Comms * MyManager::ClearCommsId(const PString &memberId) { @@ -608,13 +614,20 @@ return c; } - c = (Comms *)runningComms.GetMessageFrameOn(memberId); + c = (Comms *)voipCommsProvider.GetMessageFrameOn(memberId); if (c != NULL) { PTRACE(3, "Clear comms Id on " << memberId << " for prebuilt, ready to go frames"); return c; } + c = (Comms *)finishedCalls.GetMessageFrameOn(memberId); + if (c != NULL) { + PTRACE(3, "Clear comms Id on " << memberId + << " for finsiehd calls all done"); + return c; + } + return NULL; } @@ -636,7 +649,7 @@ Comms *c; PINDEX attempt = 0; while (attempt < 10) { - c = runningComms.GetNextComms(); + c = voipCommsProvider.GetNextComms(); if (c != NULL) break; attempt++; @@ -703,13 +716,10 @@ { /* Stop the building of new Comms instances */ PTRACE(3, "Manager\tRemove All Calls Start"); - runningComms.Terminate(); + voipCommsProvider.StopWorkerNow(); nodesMakingCalls.TerminateAllMessageFrames(); - - runningComms.WaitForTermination(); - runningComms.TerminateAllMessageFrames(); - runningComms.RemoveAllMessageFrames(); + voipCommsProvider.TerminateAllMessageFrames(); PTRACE(3, "Manager\tRemove all calls who are making calls"); /**Now we can remove all calls in the list */ @@ -720,18 +730,9 @@ } } - -void MyManager::EndEverything() -{ - syncPoint.Signal(); -} - void MyManager::Main(PArgList & /*args*/) { - syncPoint.Wait(); - - PTRACE(3, "MyManager::Main() has started the end process. Close all calls."); - supervisor.WaitForTermination(); + supervisor.WaitForEnd(); PTRACE(3, "End of supervisor thread "); for(PINDEX i = 0; i < 2; i++) { @@ -739,26 +740,34 @@ PTRACE(3, "END of all calls "); } - while(runningComms.GetSize() > 0) { - PTRACE(3, "Manager\tWait for runningComs to reach 0. Curently " - << runningComms.GetSize() << " elements"); - PThread::Sleep(100); - - PTRACE(3, "Running comms has a size of " << runningComms.GetSize()); - - while (nodesMakingCalls.GetSize() > 0) { + bool lookAgain = true; + while (lookAgain) { + lookAgain = false; + if(voipCommsProvider.GetSize() > 0) { + PTRACE(3, "Manager\tWait for runningComms to reach 0. Curently " + << voipCommsProvider.GetSize() << " elements"); + voipCommsProvider.RemoveDeadEntries(); + PThread::Sleep(100); + PTRACE(3, "Running comms has a size of " << voipCommsProvider.GetSize()); + lookAgain = true; + } + + if(nodesMakingCalls.GetSize() > 0) { PTRACE(3, "Manager\tWait for Nodes Makeing Calls to get to zero size"); + nodesMakingCalls.RemoveDeadEntries(); PThread::Sleep(100); + PTRACE(3, "nodes making calls has a size of " + << nodesMakingCalls.GetSize()); + lookAgain = true; } - PTRACE(3, "nodes making calls has a size of " << nodesMakingCalls.GetSize()); - while (activeCalls.GetSize() > 0) { + if(activeCalls.GetSize() > 0) { PTRACE(3, "Manager\tWait for active calls to get to zero size"); - PThread::Sleep(100); - } - PTRACE(3, "active calls has a size of " << activeCalls.GetSize()); + PThread::Sleep(100); + PTRACE(3, "active calls has a size of " << activeCalls.GetSize()); + lookAgain = true; + } } - PTRACE(3, "Now, we have to leave a wait period"); PThread::Sleep(2000); @@ -871,7 +880,7 @@ { PTRACE(3, "Manager\tCreate a call with any CallID"); Comms *c; - c = runningComms.GetNextComms(); + c = voipCommsProvider.GetNextComms(); if (c == NULL) { PTRACE(3, "Failed to make a new node"); @@ -909,7 +918,7 @@ if (c == NULL) { PTRACE(3, "No valid comms structure yet, so create one on the fly "); - c = runningComms.GetNextComms(); + c = voipCommsProvider.GetNextComms(); if (c == NULL) { PTRACE(1, "BAD vibes, failed to creaate comms instance "); Modified: sofaswitch/trunk/voip/mymanager.h ============================================================================== --- sofaswitch/trunk/voip/mymanager.h (original) +++ sofaswitch/trunk/voip/mymanager.h Tue Sep 30 22:35:22 2008 @@ -73,9 +73,6 @@ /**destructor */ ~MyManager(); - /**Set the syncpoint to run, which ends the main thread*/ - void EndEverything(); - /**Given the command line args, set this class up */ bool Initialise(PArgList & args); @@ -214,24 +211,23 @@ const PString & requester); /**Check the different lists of Comms instances in this Manager - (runningComms, deadComms, and nodesMakingCalls) and find the - matching Comms instance. Return a pointer to the Comms - instance (or NULL) depending on what is found */ + (runningComms and nodesMakingCalls) and find the matching Comms + instance. Return a pointer to the Comms instance (or NULL) + depending on what is found */ Comms * ClearCommsId(const PString & _memberId); /**Go through the lists of available calls, and tell them to "die now" */ void RemoveAllCalls(); - + /**This comms instance was used in a call, and the call is finished. Put this comms instance in a list. This method will then invoke a CommsConsumer class to remove this instance */ - void AppendFinishedComms(Comms *commsPtr); + void CallHasFinished(Comms *commsPtr); /**Report, in a sorted string list, the sofa id of all running comms instances in a call */ - PSortedStringList GetCurrentNames(); - + void GetCurrentNames(PSortedStringList & answer); /**Return the pointer to the iax2 endpoint. This is used for retrieving stats, and is dangerous to be used for other @@ -242,6 +238,7 @@ void *GetIax2Ep() { return NULL; } #endif + protected: /**The interface to the external script which runs python, to provide comms for us */ LaunchNewId &lid; @@ -251,7 +248,7 @@ /**A list of Comms instances, which are already built, connected to the remote server, and ready to be used */ - CommsProvider &runningComms; + CommsProvider & voipCommsProvider; /**A list of the nodes who are initiating calls through us. Frames will be deleted from this list if they have not been claimed. @@ -259,10 +256,10 @@ deleted */ MessageFrameListPending nodesMakingCalls; - /**A list of Comms instances that are known to have died. Comms - instances will be removed from this list by the CommsConsumer - class.*/ - MessageFrameList deadComms; + /**List of frames that have completed a call. They are put here, + so the cleaning mechanism of Manager.ClearCommsId() will find + them for deleteion */ + MessageFrameList finishedCalls; /**The sound system endpoint that we use to control talk to the remote sofa switch nodes */ @@ -289,9 +286,6 @@ the server */ PString ourId; - /**Sync point that can be called to end the whole thing */ - PSyncPoint syncPoint; - /**The allocation number used in the last memory leak test request*/ DWORD allocationNumber; }; Modified: sofaswitch/trunk/voip/supervisor.cxx ============================================================================== --- sofaswitch/trunk/voip/supervisor.cxx (original) +++ sofaswitch/trunk/voip/supervisor.cxx Tue Sep 30 22:35:22 2008 @@ -68,6 +68,13 @@ void Supervisor::OnTermination() { PTRACE(3, "Supervisor thread has ended"); + terminationFlag.Signal(); +} + +void Supervisor::WaitForEnd() +{ + terminationFlag.Wait(); + PTRACE(3, "MANAGER END NOW, as we discover the supervisor has stopped"); } void Supervisor::OnRunning() @@ -75,6 +82,7 @@ PTRACE(3, "Create a supervisor all done"); PTRACE(3, "Supervisor initialisation done"); PTRACE(3, "Supervisor wait for commands now"); + SetThreadName(PString("Supervisor. ") + GetOurId()); #ifdef PMEMORY_CHECK PMemoryHeap::GetState(memoryState); @@ -133,14 +141,6 @@ { SetRemoteController(sender); PTRACE(3, "Supervisor says::End everything now"); - VoipProcess::Current().GetManager()->EndEverything(); - PTRACE(3, "Supervisor says::End everything now, and it has been initiated."); -} - -void Supervisor::SilentlyEndAll() -{ - PTRACE(3, "Supervisor says::End everything now"); - VoipProcess::Current().GetManager()->EndEverything(); } void Supervisor::ProcessHangup(const PStringArray &cmds, const PString & sender) Modified: sofaswitch/trunk/voip/supervisor.h ============================================================================== --- sofaswitch/trunk/voip/supervisor.h (original) +++ sofaswitch/trunk/voip/supervisor.h Tue Sep 30 22:35:22 2008 @@ -73,6 +73,10 @@ /**Report here the extra commands the voip supervisor can process */ virtual PStringArray GetDescendantCommands() { return extraAvailableCommands; } + /**Wait in here for the termination syncpoint to fire. Only one + thread should wait in here. The syncpoint will only fire + once. */ + void WaitForEnd(); protected: /**The supervisor has completed. This method is called when this @@ -121,10 +125,6 @@ exit the program. This is the supervisor, so we can cause a shutdown. */ virtual bool OkToQuitNow() { return true; } - /**We have a comms error, so just close everything right now. Send - no message to the remote node - just end end end. */ - virtual void SilentlyEndAll(); - /**The list of extra commands we can handle in the voip supervisor */ PStringArray extraAvailableCommands; @@ -133,6 +133,9 @@ /**The interface to the external script which runs python, to provide XMPP comms for us */ LaunchNewId &launchNewId; + /**The sync point that fires ONCE, to mark the end of this thread */ + PSyncPoint terminationFlag; + /**Internally and seriously protected structure to handle the list of preallocated ids */ // CommsProvider availableComms; Modified: sofaswitch/trunk/voip/udpconnection.cxx ============================================================================== --- sofaswitch/trunk/voip/udpconnection.cxx (original) +++ sofaswitch/trunk/voip/udpconnection.cxx Tue Sep 30 22:35:22 2008 @@ -177,49 +177,12 @@ return formats; } - -OpalMediaStream * OpalUdpSsConnection::CreateMediaStream( - const OpalMediaFormat & mediaFormat, - unsigned sessionID, - bool isSource) -{ - PTRACE(3, "Sofa\tCreateMediaStream " << mediaFormat); - if (sessionID != OpalMediaFormat::DefaultAudioSessionID) { - PTRACE(3, "Sofa\tCreateMediaStream " << mediaFormat << " default one"); - return OpalConnection::CreateMediaStream(mediaFormat, sessionID, isSource); - } - - PSoundChannel * soundChannel = CreateSoundChannel(isSource); - if (soundChannel == NULL) - return NULL; - - return new OpalAudioMediaStream(*this, mediaFormat, sessionID, isSource, 2, soundChannel); -} - void OpalUdpSsConnection::CloseMediaStreams() { PTRACE(5, "OpalUdpSsConnection::CloseMediaStreams"); OpalConnection::CloseMediaStreams(); } -bool OpalUdpSsConnection::OnOpenMediaStream(OpalMediaStream & mediaStream) -{ - PTRACE(3, "OnOpenMediaStream for " << mediaStream); - if (!OpalConnection::OnOpenMediaStream(mediaStream)) - return false; - - if (mediaStream.IsSource()) { - OpalMediaPatch * patch = mediaStream.GetPatch(); - if (patch != NULL && - (mediaStream.GetSessionID() == OpalMediaFormat::DefaultAudioSessionID)) { - silenceDetector->SetParameters(endpoint.GetManager().GetSilenceDetectParams()); - patch->AddFilter(silenceDetector->GetReceiveHandler(), OpalPCM16); - } - } - - return true; -} - void OpalUdpSsConnection::Release(CallEndReason reason) { ((MyOpalCall &)(GetCall())).CloseThisId(); @@ -244,6 +207,22 @@ ((MyOpalCall &)GetCall()).SendThisMessageTo(message, destination); } + +OpalMediaStream * OpalUdpSsConnection::CreateMediaStream(const OpalMediaFormat & mediaFormat, + unsigned sessionID, + PBoolean isSource) +{ + if (mediaFormat.GetMediaType() != OpalMediaType::Audio()) + return OpalConnection::CreateMediaStream(mediaFormat, sessionID, isSource); + + PSoundChannel * soundChannel = CreateSoundChannel(isSource); + if (soundChannel == NULL) + return NULL; + + return new OpalAudioMediaStream(*this, mediaFormat, sessionID, isSource, 2, soundChannel); +} + + PSoundChannel * OpalUdpSsConnection::CreateSoundChannel(bool isSource) { PTRACE(1, "Create UDP Sound Channel "); Modified: sofaswitch/trunk/voip/udpconnection.h ============================================================================== --- sofaswitch/trunk/voip/udpconnection.h (original) +++ sofaswitch/trunk/voip/udpconnection.h Tue Sep 30 22:35:22 2008 @@ -104,6 +104,26 @@ bool withMedia /// Open media with alerting ); + /**Open a new media stream. + This will create a media stream of an appropriate subclass as required + by the underlying connection protocol. For instance H.323 would create + an OpalRTPStream. + + The sessionID parameter may not be needed by a particular media stream + and may be ignored. In the case of an OpalRTPStream it us used. + + Note that media streams may be created internally to the underlying + protocol. This function is not the only way a stream can come into + existance. + + The default behaviour is pure. + */ + virtual OpalMediaStream * CreateMediaStream( + const OpalMediaFormat & mediaFormat, ///< Media format for stream + unsigned sessionID, ///< Session number for stream + PBoolean isSource ///< Is a source stream + ); + /**Called at the end of the call setup process. Audio is happening now */ virtual void OnEstablished(); @@ -130,41 +150,6 @@ */ virtual OpalMediaFormatList GetMediaFormats() const; - /**Open a new media stream. - This will create a media stream of an appropriate subclass as required - by the underlying connection protocol. For instance H.323 would create - an OpalRTPStream. - - The sessionID parameter may not be needed by a particular media stream - and may be ignored. In the case of an OpalRTPStream it us used. - - Note that media streams may be created internally to the underlying - protocol. This function is not the only way a stream can come into - existance. - - The default behaviour is pure. - */ - virtual OpalMediaStream * CreateMediaStream( - const OpalMediaFormat & mediaFormat, /// Media format for stream - unsigned sessionID, /// Session number for stream - bool isSource /// Is a source stream - ); - - /**Call back when opening a media stream. - This function is called when a connection has created a new media - stream according to the logic of its underlying protocol. - - The usual requirement is that media streams are created on all other - connections participating in the call and all of the media streams are - attached to an instance of an OpalMediaPatch object that will read from - one of the media streams passing data to the other media streams. - - The default behaviour calls the ancestor and adds a PCM silence - detector filter. - */ - virtual bool OnOpenMediaStream( - OpalMediaStream & stream /// New media stream being opened - ); /**Send a user input indication to the remote endpoint. Modified: sofaswitch/trunk/voip/udpsound.cxx ============================================================================== --- sofaswitch/trunk/voip/udpsound.cxx (original) +++ sofaswitch/trunk/voip/udpsound.cxx Tue Sep 30 22:35:22 2008 @@ -100,6 +100,7 @@ bool UdpSoundChannel::Write(const void * buffer, PINDEX amount) { + PTRACE(4, "MEDIA\tIOWrite"); { WORD dPort; PIPSocket::Address destination; @@ -147,6 +148,7 @@ bool UdpSoundChannel::Read(void * buffer, PINDEX amount) { + PTRACE(4, "MEDIA\tIORead"); if (!IsOpen()) { PTRACE(3, "MEDIA\tRead failed as this sound channel is closed"); return false;