Node - Mediator Interface Definitions - #6
Conversation
| * @param host socket host | ||
| * @param port socket port | ||
| */ | ||
| void removeSubscriber(std::string const &topic_name, std::string const &host, int const port); |
There was a problem hiding this comment.
I think this maybe makes sense to call from the destructor of subscriber? But then subscriber has to internally make a call to a node to get node to call this callback on mediator over rpc since only node is connected to the mediator. There's probably two ways we could go:
- Since there isn't a delete_subscriber() function for node the subscriber will only be destroyed when the node is destroyed. We could demote this to a helper function that gets called by remove_node() callback you made below. As the node dies it calls the remove node callback on mediator and the mediator then looks up the publishers and subscribers associated with that node and calls remove publisher and remove subscriber on them.
The user can't delete the publisher's and subscriber's individually because it's a weak_ptr or something right? The publishers and subscribers are owned by node?
- We add a delete_subscriber() function to the Node interface. Then this makes sense and gets called internally by the delete_subscriber function over rpc.
Whichever seems best to you, or if you have another way.
There was a problem hiding this comment.
Definitely should be called in the dtor of sub
There was a problem hiding this comment.
After reading the whole codebase again I really don't agree with this. I think our implementation will be much simpler if we just rely on Node getting destroyed to call the callbacks that remove the appropriate publishers and subscribers. The Publisher and Subscriber have life cycles only can end when Node is destroyed so it doesn't make sense to separate these into entirely different callbacks. Keeping them as helper functions is good for compartmentalization though.
| * @param host socket host | ||
| * @param port socket port | ||
| */ | ||
| void removePublisher(std::string const &topic_name, std::string const &host, int const port); |
There was a problem hiding this comment.
Exactly the same comment as above about removePublisher().
| * @param host socket host | ||
| * @param port socket port | ||
| */ | ||
| void removeNode(std::string const &node_name, std::string const &host, int const port); |
There was a problem hiding this comment.
Gets registered as the closing callback for the rpc connection to each Node?
If the Node gets killed locally, it closes the rpc connection and the mediator automatically calls this removeNode callback.
If the Mediator decides to kill the node, then the mediator closes the rpc connection and which automatically calls this removeNode callback and the Node's closing callback sequence.
There was a problem hiding this comment.
Question that applies to all of the above callbacks:
Should we also have little wrapper functions for each of these callbacks that handle parsing the string and int arguments from a single JSON object using keywords?
To actually register these as callbacks for the JsonRPCSocket's they need to just take one JSON object as input. I think the way you have it right now is way more readable so maybe just little wrapper functions that take a JSON as a parameter then parse out std::string node_name, std::string host, int port, etc? Could call it like removeNodeJson, addSubscriberJson, or something
| * Converts URI to name of node | ||
| */ | ||
| std::unordered_map<std::string, std::string> nodeTable_; | ||
|
|
There was a problem hiding this comment.
If we wanted to have the removeNode() callback also remove publishers and subscribers, we would need a node URI to publisher and subsciber URI map or maps.
Or we could make nodeTable map to a struct with the node's name, a vector of publisher URI's, and a vector of subscriber URI's.
Only necessary if we do option 1 mentioned above next to removeNode() function.
There was a problem hiding this comment.
We definitely also need a vector of Node URI's mapping shared_ptr to each other somewhere. We need to know which rpc socket talks to which node.
If we edit nodeTable to store a struct of stuff as the value for the unordered_map, we could add the pointer to the rpc socket in there as well. That would probably make the most sense since there's a lot of different information we might want to know about a Node.
- name
- publisher URI's
- subscriber URI's
- ptr to connection json rpc socket
| /** | ||
| * Created listener thread for RPC calls | ||
| */ | ||
| void RPCListenerThread(); |
There was a problem hiding this comment.
This is where the ServerSocket is accepting incoming connections made by Node's JsonRPCClientSocket, right?
We accept the connection, setup the appropriate callbacks (that you put below) for the resulting ConnectionSocket, and then start up the RPC for that socket and store the shared_ptr to the RPC socket so we can talk to that node in the future.
| static std::string messageName() { | ||
| // return the exact name of the struct | ||
| // idk how I want to handle namespaces | ||
| // i.e. Messages::String |
There was a problem hiding this comment.
Ahhhhh the message interface is so tricky.
Maybe we want to have an abstract base message class?
- Pure virtual function to initialize a the message from a json
- Pure virtual function to generate a json from the content of the message (Both would use that cool macro on line 37)
Then to make different message types the users of the library just inherit from the base message class and define all the message member variables they want. They are forced to define to and from Json functions because of the pure virtual functions, so we can always rely on them being there in other pieces of code.
That's the simplest thing I can think of for making these a little more uniform. I'll open this as a maybe for later, low priority issue.
I think having a Messages:: namespace seems like a good idea for sure though
| /** | ||
| * Thread pool as needed | ||
| */ | ||
| // std::vector<std::thread> threadPool_; |
There was a problem hiding this comment.
Need a single shared_ptr in here somewhere to maintain the connection to mediator node.
If it is private, then we won't be able to call callbacks like mediator's removeSubscriber/Publisher from the destructor of Publiser or Subscriber and we'll need to change the way that callback works on mediator.
Can't make it public because then users can just talk to mediator unchecked.
Making Publisher and Subscriber friends would work it's just overly coupled to have friends both ways. If Publisher and Subscriber just rely on Node's destruction to call all the remove callbacks then the json rpc socket can just be private.
| CallbackT &&callback); | ||
|
|
||
| /** | ||
| * Creator of subscribers |
| /** | ||
| * Continuously handles incoming information such as new publishers and subscribed messages | ||
| */ | ||
| void spin(); |
There was a problem hiding this comment.
Are we still doing the spin() interface here? I don't think it makes sense. We just have a JsonRPCSocket that is always connected to mediator. The JsonRPCSocket's internal receiving thread just handles all the communication with Mediator. I think we just need one more thread internal to Node to watch the SignalHandler using that status function you defined below.
|
|
||
| /** | ||
| * Thread pool as needed | ||
| */ |
There was a problem hiding this comment.
Depending on how much work publishers and subscribers's do with their own internal threads we may not need a whole pool here. Oh, that's probably why you commented it out.
| /** | ||
| * TODO | ||
| */ | ||
| void socketListener(); |
There was a problem hiding this comment.
We need a ServerSocket that subscribers can connect to. This function should spin up the thread on line 102 and call accept on the ServerSocket and stick the resulting ConnectionBsonSocket's in the subscribers_ vector in a loop.
| * Creates thread to receive incoming messages on one file descriptor | ||
| * @param socket_fd File descriptor | ||
| */ | ||
| void receiverThread(int socket_fd); |
There was a problem hiding this comment.
We need a vector of ptr's to BsonClientSockets somewhere in this class. We are trying to not work with raw file descriptors anymore.
We agreed one a few things about the subscriber architecture:
- that we would have one thread per socket connected to a publisher
- each one of these threads receives on the socket and calls the registered callback
- multiple threads can call the callback concurrently
I implemented the message socket interface so that both sendMessage and receiveMessage will throw errors on closed sockets so we don't need to directly track publishers or subscribers being destroyed. If the bson socket for a given publisher throws PeerClosedException, we should just remove that socket from the vector and join the thread that was calling on it. I think we can rely on the Mediator getting updated appropriately by the totally separate callback mechanism you set up.
There was a problem hiding this comment.
This also makes the mediator implementation simpler because when a node gets removed we just assume subscribers will figure it out appropriately via their receiving threads getting PeerSocketClosed thrown.
| /** | ||
| * Lookup socket file descriptor -> thread. One to one | ||
| */ | ||
| std::unordered_map<int, std::thread> publisherListenerThreads_; |
There was a problem hiding this comment.
Because the shutdown sequence for BsonSocket just relies on throwing we probably won't ever need to look up threads. We just start a new thread when we hear about a new publisher, and it closes either when we kill the node or when the bson socket throws PeerClosedException.
| /** | ||
| * number of messages to be allowed on the queue | ||
| */ | ||
| std::uint32_t queue_size_; |
There was a problem hiding this comment.
Where is the queue whose size we are tracking?
A couple options here:
- Add a queue size interface to bson socket and maintain this queue size for every bson socket.
- Add a single queue of messages that all the sockets pass their messages to. All sockets threads dump to this queue then there is a single calling thread.
- Drop the support for keeping a queue size.
Do you have an opinion on this?
| /** | ||
| * Conidition variable that locks receiving messages until new publisher is added to list and connection is created | ||
| */ | ||
| std::condition_variable spin_cv; //Broadcasted when spin() or spinOnce() is called. Blocks all receivers. |
There was a problem hiding this comment.
This gets paired with publisherListenerThreadsMutex?
I think we may need a condition variable and a mutex for every thread that is receiving on a socket. If each thread needs to wait on the condition, then each thread needs its own lock.
I think this is probably the advantage of receiving sockets reading into a single big queue and having one thread that is calling the callback and gets controlled by the condition of whether or not spin() has been called.
| /** | ||
| * Mutex for publisherListenerThreads_ to prevent race conditions | ||
| */ | ||
| std::mutex publisherListenerThreadsMutex_; |
There was a problem hiding this comment.
See comment at condition variable
| * Number of threads waiting | ||
| * TODO: More information | ||
| */ | ||
| int numThreadsWaiting_ = 0; |
There was a problem hiding this comment.
Yeah not sure either. Considering Tom isn't on this anymore and didn't actually comment out his ideas we should probably just remove it.
| /** | ||
| * TODO: Unsure | ||
| */ | ||
| bool listenerThreadsCondition = false; |
2 participants
Writing up header file comments to nail down communication between the node and the mediator.