Author: derek Date: Fri Oct 27 19:40:34 2006 New Revision: 150 Modified: sofaswitch/trunk/mcu/manager.cxx sofaswitch/trunk/mcu/manager.h sofaswitch/trunk/mcu/member.cxx sofaswitch/trunk/mcu/member.h sofaswitch/trunk/mcu/participant.cxx sofaswitch/trunk/mcu/participant.h Log: Add some better handling of the members in a participant dict. Better use of PSafePtrs. More testing required. Modified: sofaswitch/trunk/mcu/manager.cxx ============================================================================== --- sofaswitch/trunk/mcu/manager.cxx (original) +++ sofaswitch/trunk/mcu/manager.cxx Fri Oct 27 19:40:34 2006 @@ -173,7 +173,7 @@ supervisor.Resume(); } -void Manager::RemoveMemberId(PString memberId) +void Manager::RemoveMemberId(const PString & memberId) { PTRACE(3, "Remove MemberID \"" << memberId << "\" START"); { @@ -189,7 +189,7 @@ } -void Manager::RemoveConferenceRoomId(PString conferenceRoomId) +void Manager::RemoveConferenceRoomId(const PString & conferenceRoomId) { PTRACE(3, "Remove conference room id from list of conference rooms " << conferenceRoomId); activeRooms.RemoveAt(conferenceRoomId); Modified: sofaswitch/trunk/mcu/manager.h ============================================================================== --- sofaswitch/trunk/mcu/manager.h (original) +++ sofaswitch/trunk/mcu/manager.h Fri Oct 27 19:40:34 2006 @@ -99,11 +99,11 @@ /**Cause the PSafePtr<Member> of the specified id to be removed from the * activeMembers list in a threadsafe fashion */ - void RemoveMemberId(PString memberId); + void RemoveMemberId(const PString & memberId); /**Cause the PSafePtr<ConferenceRoom> of the specified id to be * removed from the activeRooms list in a threadsafe fashion */ - void RemoveConferenceRoomId(PString roomId); + void RemoveConferenceRoomId(const PString & roomId); /**The entry point, where the main control loop sits */ void Main(); Modified: sofaswitch/trunk/mcu/member.cxx ============================================================================== --- sofaswitch/trunk/mcu/member.cxx (original) +++ sofaswitch/trunk/mcu/member.cxx Fri Oct 27 19:40:34 2006 @@ -78,7 +78,6 @@ IncCountMembers(); PTRACE(3, "Member\tConstructor " << *this); - participants.SetAutoDeleteObjects(); keepGoing = TRUE; bufferSize = PCM_BUFFER_SIZE; @@ -129,15 +128,9 @@ return duration; } -void Member::AddMemberToRoom(PString newMember) +void Member::AddMemberToRoom(const PString & newMember) { - PSafePtr <Member> member = manager.GetMember(newMember); - - if (member != NULL) { - Participant *participant = new Participant(manager, memberId, member); - participants.AddNewParticipant(participant); - PTRACE(3, "Finished adding " << newMember << " to " << memberId); - } + participants.AddNewParticipant(manager, memberId, newMember); } @@ -159,7 +152,7 @@ void Member::DestroyThisParticipant(const PString &partyKey) { PTRACE(3, "Member " << memberId << " is now to remove participant " << partyKey); - participants.RemoveAt(partyKey); + participants.RemoveParticipant(partyKey); } Modified: sofaswitch/trunk/mcu/member.h ============================================================================== --- sofaswitch/trunk/mcu/member.h (original) +++ sofaswitch/trunk/mcu/member.h Fri Oct 27 19:40:34 2006 @@ -89,7 +89,7 @@ /**A new member has joined this conference room. This method is then called to add the new member to this Members's list of other members */ - void AddMemberToRoom(PString newMember); + void AddMemberToRoom(const PString & newMember); /**Remove this member from the conference room*/ void RemoveFromRoom(); Modified: sofaswitch/trunk/mcu/participant.cxx ============================================================================== --- sofaswitch/trunk/mcu/participant.cxx (original) +++ sofaswitch/trunk/mcu/participant.cxx Fri Oct 27 19:40:34 2006 @@ -39,26 +39,31 @@ //////////////////////////////////////////////////////////////////////////////// -Participant::Participant(Manager & _mgr, const PString &baseMemberId, PSafePtr <Member> member) +Participant::Participant(Manager & _mgr, ParticipantDict & _partyDict, + const PString &baseMemberId, + PSafePtr <Member> member) : manager(_mgr), - sourceMemberId(baseMemberId) + dictionary(_partyDict), + srcMemberId(baseMemberId) { - memberId = member->GetMemberId(); - SetKey (baseMemberId + PString("__") + memberId); - SetMatchingKey(memberId + PString("__") + baseMemberId); + dstMemberId = member->GetMemberId(); + SetKey (srcMemberId + PString("-->") + dstMemberId); + SetMatchingKey(dstMemberId + PString("-->") + srcMemberId); + isClearing = FALSE; memberPtr = member; PTRACE(3, "Participant\tConstructor " << *this); - memberPtr->SafeReference(); currentReadIndex = 0; IncCountParticipants(); + dictionary.ParticipantCreated(); } Participant::~Participant() { - DecCountParticipants(); - PTRACE(3, "Participant\tDestructor " << *this); - new MatchingParticipantConsumer(memberPtr, GetMatchingKey()); + DecCountParticipants(); + + PTRACE(3, "Participant " << *this << " destroyed dead and gone"); + dictionary.ParticipantDestroyed(); } void Participant::PrintOn(ostream & strm /// Stream to output text representation @@ -69,7 +74,7 @@ PString Participant::GetMemberId() { - return memberId; + return srcMemberId; } BOOL Participant::ReadAndMixAudio(BYTE * data, PINDEX amount) @@ -77,8 +82,52 @@ return memberPtr->ReadAndMixAudio(data, amount, currentReadIndex); } +void Participant::OnReleased() +{ + dictionary.ParticipantReleased(GetKey()); +} + + +void Participant::Clear() +{ + PTRACE(3, *this << " has been told to CLEAR"); + { + PWaitAndSignal m(clearMutex); + if (isClearing) + return; + isClearing = TRUE; + } + // Add a reference for the thread we are about to start + SafeReference(); + PString clearName = PString("OnRelease ") + GetKey(); + PThread::Create(PCREATE_NOTIFIER(OnReleaseThreadMain), 30000, + PThread::AutoDeleteThread, + PThread::NormalPriority, + clearName); +} + + +void Participant::OnReleaseThreadMain(PThread &, INT) +{ + PTRACE(3, "OnRelease thread started for " << *this); + + DestroyNeighboursLink(); + + OnReleased(); + + PTRACE(3, "Participant OnRelease thread completed for " << *this); + // Dereference on the way out of the thread + SafeDereference(); +} + +void Participant::DestroyNeighboursLink() +{ + new MatchingParticipantConsumer(manager, dstMemberId, GetMatchingKey()); +} + + //////////////////////////////////////////////////////////////////////////////// -void ParticipantDict::DeleteObject(PObject * object) const +void ParticipantD::DeleteObject(PObject * object) const { Participant * oneParticipant = PDownCast(Participant, object); if (oneParticipant != NULL) { @@ -87,6 +136,58 @@ } } +///////////////////////////////////////////////////////////////////////////// +ParticipantDict::ParticipantDict() +{ + countEntries = 0; + SetAutoDeleteObjects(); + + shuttingDown = FALSE; + PString name = psprintf("garbage %x", this); + PThread::Create(PCREATE_NOTIFIER(GarbageMain), 0, + PThread::AutoDeleteThread, + PThread::LowPriority, + name); +} + +ParticipantDict::~ParticipantDict() +{ + WaitForTotallyEmpty(); + + garbageCollectExit.Signal(); + garbageFinished.Wait(); +} + + +void ParticipantDict::GarbageMain(PThread &, INT) +{ + while (!garbageCollectExit.Wait(500)) + GarbageCollection(); + + garbageFinished.Signal(); +} + + +void ParticipantDict::GarbageCollection() +{ + DeleteObjectsToBeRemoved(); + PTRACE(5, "Garbage on ParticipantDict done, remaining, " << GetSize()); +} + +void ParticipantDict::WaitForTotallyEmpty() +{ + while (countEntries > 0) { + PThread::Sleep(500); + } +} + +void ParticipantDict::PrintOn(ostream & strm /// Stream to output text representation + ) const +{ + for (PSafePtr<Participant> participant(*this, PSafeReadOnly); participant != NULL; ++participant) + strm << participant->GetKey() << " "; +} + PSortedStringList ParticipantDict::GetConferenceParticipants() { PSortedStringList res; @@ -99,20 +200,24 @@ void ParticipantDict::ReadAndMixAudio(BYTE *data, PINDEX amount) { - for (PSafePtr<Participant> participant(*this, PSafeReadOnly); participant != NULL; ++participant) { - BOOL readStatus = participant->ReadAndMixAudio(data, amount); - if (!readStatus) { - PTRACE(3, "Remove participant from list " << *participant); - RemoveAt(participant->GetKey()); + for (PSafePtr<Participant> participant(*this, PSafeReadOnly); participant != NULL; ++participant) { + BOOL readStatus = participant->ReadAndMixAudio(data, amount); + if (!readStatus) + participant->Clear(); } - } } - -void ParticipantDict::AddNewParticipant(PSafePtr<Participant> src) +void ParticipantDict::AddNewParticipant(Manager & mgr, + const PString & srcMember, + const PString & dstMember) { - PTRACE(3, "Add new participant of " << src->GetMemberId()); - SetAt(src->GetKey(), src); + PSafePtr <Member> dstMemberPtr = mgr.GetMember(dstMember); + if (dstMemberPtr != NULL) { + PSafePtr <Participant> participant = new Participant(mgr, *this, srcMember, dstMemberPtr); + SetAt(participant->GetKey(), participant); + + PTRACE(3, "Done add new participant of " << participant->GetKey()); + } } void ParticipantDict::EmptyParticipantDict() @@ -127,44 +232,51 @@ } -void ParticipantDict::RemoveParticipant(PString member) +void ParticipantDict::RemoveParticipant(const PString & member) { PTRACE(3, "RemoveParticipant for " << member); - for (PSafePtr<Participant> participant(*this, PSafeReadWrite); participant != NULL; ++participant) { - if (participant->GetMemberId() == member) { - PString tempKey = participant->GetKey(); - PTRACE(3, "Start Remove participant from list " << member << " and " << tempKey); - RemoveAt(participant->GetKey()); - PTRACE(3, "Finished Remove participant from list " << member << " and " << tempKey); - return; + PSafePtr<Participant> participant = FindWithLock(member, PSafeReadOnly); + if (participant != NULL) + participant->Clear(); +} + +void ParticipantDict::ParticipantReleased(const PString & partKey) +{ + PSafePtr<Participant> participant = FindWithLock(partKey, PSafeReadWrite); + if (participant != NULL) { + PTRACE(3, "Participant OnReleased " << partKey << ".. Now do RemoveAt()"); + RemoveAt(partKey); } - PTRACE(3, "Did not find participant " << member << " in list"); - } } ///////////////////////////////////////////////////////////////////////////// -MatchingParticipantConsumer::MatchingParticipantConsumer(Member *m, const PString & _key) +MatchingParticipantConsumer::MatchingParticipantConsumer(Manager & _mgr, + const PString & _otherMember, + const PString & _partyKey) :PThread(10000, AutoDeleteThread, LowPriority, "%X participantdelete"), - member(m), key(_key) + mgr(_mgr), + otherMember(_otherMember), + partyKey(_partyKey) { - PTRACE(3, "ParticipantConsumer contructor " << key); + PTRACE(3, "ParticipantConsumer contructor " << partyKey); Resume(); } MatchingParticipantConsumer::~MatchingParticipantConsumer() { - PTRACE(3, "ParticipantConsumer destructor " << key); + PTRACE(3, "ParticipantConsumer destructor " << partyKey); } void MatchingParticipantConsumer::Main() { - if (member == NULL) - return; + PSafePtr <Member> member = mgr.GetMember(otherMember); + + if (member == NULL) + return; - member->DestroyThisParticipant(key); - member->SafeDereference(); + member->DestroyThisParticipant(partyKey); } Modified: sofaswitch/trunk/mcu/participant.h ============================================================================== --- sofaswitch/trunk/mcu/participant.h (original) +++ sofaswitch/trunk/mcu/participant.h Fri Oct 27 19:40:34 2006 @@ -40,32 +40,51 @@ //////////////////////////////////////////////////////////////////////////////// - /**The thread safe list of Participants to each conference */ -class ParticipantDict : public PSafeDictionary<PString, Participant> +class ParticipantD : public PSafeDictionary<PString, Participant> { /**One function that we have to define for this usage, and that is the deletion of existing members. PWLib will have removed the object to be deleted from the list. We simply need to delete it. */ virtual void DeleteObject(PObject * object) const; - +}; +///////////////////////////////////////////////////////////////////////////// + +/**The thread safe list of Participants to each conference. This class is + descended off ParticipantD, and gives us a chance to put a constructor + in. */ +class ParticipantDict : ParticipantD +{ public: - /**Step through the list, and do a ReadAndMix audio process on each of - them. */ - void ReadAndMixAudio(BYTE *data, PINDEX amount); + /**Constructor, which creates the garbage cleaning thread, and turns on + auto deleteion of objects */ + ParticipantDict(); - /**Add an instance of the Participant class (supplied as an argument) - to the end of this list*/ - void AddNewParticipant(PSafePtr<Participant> src); + /**Destructor. Wait for this dictionary to be totally empty, and then + stop the garbage collection,*/ + ~ParticipantDict(); - /**Remove the participant matching the specified member id*/ - void RemoveParticipant(PString _memberId); - - /**Get the names of the members in this conference */ - PSortedStringList GetConferenceParticipants(); - - /**Tell each of the members in this list to go away and die. When + /**pretty print information in this class. It lists all the + current keys in this class. */ + virtual void PrintOn(ostream & strm /// Stream to output text representation + ) const; + + /**Step through the list, and do a ReadAndMix audio process on each of + them. */ + void ReadAndMixAudio(BYTE *data, PINDEX amount); + + /**Add an instance of the Participant class to the end of this list. If we + cannot find the requiste member to add, just do nothing.*/ + void AddNewParticipant(Manager & mgr, const PString & srcMember, const PString & dstMember); + + /**Remove the participant matching the specified member id*/ + void RemoveParticipant(const PString & _memberId); + + /**Get the names of the members in this conference */ + PSortedStringList GetConferenceParticipants(); + + /**Tell each of the members in this list to go away and die. When each Participant is destroyed (which it will do since we are the only ones who know about a participant, the destructor of that Participant will cause the matching Participant to be destroyed. @@ -84,7 +103,57 @@ there will be no participants in this dictionary. The reason is that a Participant links from one Member to another Member. */ - void EmptyParticipantDict(); + + void EmptyParticipantDict(); + +/**Wait for this class to go totally empty, with zero entries in it. */ + void WaitForTotallyEmpty(); + + /**A particpant calls this method when the participant is + constructed. This method is used to keep track of the number of + participant instances in this dictionary. */ + void ParticipantCreated() { ++countEntries; } + + /**A particpant calls this method when the participant is + deleted. This method is used to keep track of the number of + participant instances in this dictionary. */ + void ParticipantDestroyed() { ++countEntries; } + + /**A participant has gone through it's clear process, and now needs to + advise this dictionary that it has been released */ + void ParticipantReleased(const PString & partKey); + + /**The method used to invoke the cleaning up of old Participants in this + dictionary */ + void GarbageCollection(); + + protected: + /**Mutex to protect access to the shuttingDown flag */ + PMutex addNewParticipantMutex; + + /**Flag to indicate we are closing this participant dictionary, so prevent t +he + adding of more members to this dictionary */ + BOOL shuttingDown; + + /**The flag used to time the tests of the garbage collector, and to signal + * its demise */ + PSyncPoint garbageCollectExit; + + /**The flag used to signal the demise of the garbage collector */ + PSyncPoint garbageFinished; + +#ifdef DOC_PLUS_PLUS + /**The garbage collector runs this method to do garbage collection */ + virtual void GarbageMain(PThread &, INT); +#else + PDECLARE_NOTIFIER(PThread, ParticipantDict, GarbageMain); +#endif + + private: + /**Count of the additions/deleteions to this class */ + PAtomicInteger countEntries; + }; ///////////////////////////////////////////////////////////////////////////// @@ -116,6 +185,7 @@ /**Constructor, which creates this member, who has link to the manger and knows the conference we are a part of */ Participant(Manager & _mgr, + ParticipantDict & partDict, const PString &baseMember, PSafePtr <Member> member); @@ -151,10 +221,41 @@ to and the member who created us. */ PString GetMatchingKey() const { return matchingKey; } + /**End this particular Participant, and remove it dead. Do notifiy the + ParticipantDict this Participant is stored in. This is the method to + call, if you want this Participant to die. */ + void Clear(); + protected: +#ifdef DOC_PLUS_PLUS + /**This contains the 1 notifier that is used when closing down this + instance of Participant class. It is called by a custom + thread, which deletes the class */ + virtual void OnReleaseThreadMain(PThread &, INT); +#else + PDECLARE_NOTIFIER(PThread, Participant, OnReleaseThreadMain); +#endif + + /**Clean up the termination of this Participant. This function can do any + internal cleaning up and waiting on background threads that may be + using the connection object. + + The default behaviour calls the ParticipantDict function of the same + name. + */ + virtual void OnReleased(); + + /**Destroy the link from us to the other. This causees the other to + quietly destroy its link to us, but not tell us it has died */ + void DestroyNeighboursLink(); + + /**link to the manager, who knows everything about everything */ Manager & manager; + /**Link to the participant dict who will hold this participant */ + ParticipantDict & dictionary; + /**Set the key which this participant is known by. The key describes the id of the member who created us, and the member we are connected to. The Key is set in the constructor of this class. */ @@ -169,24 +270,33 @@ PINDEX currentReadIndex; /**The remote member that we are connected to */ - Member * memberPtr; + PSafePtr <Member> memberPtr; /**The id of the Member who will use us to link to memberId */ - PString sourceMemberId; + PString srcMemberId; /**The id of the member we link to */ - PString memberId; + PString dstMemberId; /**The key by which this participant is known by. The key describes the id of the member who created us, and the member we are - connected to. The key is a concatanation of the sourceMemberId - and the memberId */ + connected to. The key is a concatanation of the srcMemberId + and the dstMemberId */ PString key; /**The m key by which the corresponding participant to us is known by. The key describes the id of the member we are connected to and the member who created us. */ PString matchingKey; + + /**Mutex to protect access to the isClearing flag, which ensures that only + one thread attempts to clear this participant */ + PMutex clearMutex; + + /**Flag to indicate we are being cleared, or have been cleared. Thus, if + two threads call Clear at the same time, only one thread is created to + do the clearing. */ + BOOL isClearing; }; ///////////////////////////////////////////////////////////////////////////// @@ -202,7 +312,9 @@ /**Constructor - create this class, which is given the class to destroy. This class then does the rest */ - MatchingParticipantConsumer(Member *m, const PString & _key); + MatchingParticipantConsumer(Manager & _mgr, + const PString &_otherMember, + const PString & _partyKey); /**Destructor. */ ~MatchingParticipantConsumer(); @@ -211,13 +323,15 @@ void Main(); protected: - + /**Reference to the manager of this conference */ + Manager & mgr; + /**The instance we have to talk to, so the matching Participant class can be deleted */ - Member *member; + PString otherMember; /**The key of the matching Participant class we want to delete */ - PString key; + PString partyKey; }; #endif // _Mcu_Participant_H