caffe-distri/src/main/cpp/util/socket.cpp
// Handle each peer connection in their own handlers
Handle each peer connection in their own handlers
Handle each peer connection in their own handlers
void *client_connection_handler(void *metadata) {
  struct connection_details * cd = (struct connection_details*) metadata;
  struct message_header mh;
  while (1) {
    receive_message_header(cd->serving_fd, &mh);
    SocketChannel* sc = cd->sa->channels->at(mh.rank).get();
    sc->serving_fd = cd->serving_fd;
    uint8_t* read_buffer = new uint8_t[mh.size];
    uint8_t* marker = read_buffer;
    int cur_cnt = 0;
    int max_buff = 0;
    while (cur_cnt < mh.size) {
      if ((mh.size - cur_cnt) > 256)
        max_buff = 256;
      else
        max_buff = mh.size - cur_cnt;
      int n = read(sc->serving_fd, marker, max_buff);
      CHECK(n >= 0) << "ERROR: Reading data from client";
      marker = marker + n;
      cur_cnt = cur_cnt + n;
    }
    QueuedMessage* mq = new QueuedMessage(mh.type,
                                          mh.size, read_buffer);
    if(mh.type == DIFF)
      sc->receive_queue.push(mq);
    else
      sc->receive_queue_ctrl.push(mq);
  }
  return NULL;
}
void * client_connection_handler ( void * metadata ) { struct connection_details * cd = ( struct connection_details * ) metadata ; struct message_header mh ; while ( 1 ) { receive_message_header ( cd -> serving_fd , & mh ) ; SocketChannel * sc = cd -> sa -> channels -> at ( mh . rank ) . get ( ) ; sc -> serving_fd = cd -> serving_fd ; uint8_t * read_buffer = new uint8_t [ mh . size ] ; uint8_t * marker = read_buffer ; int cur_cnt = 0 ; int max_buff = 0 ; while ( cur_cnt < mh . size ) { if ( ( mh . size - cur_cnt ) > 256 ) max_buff = 256 ; else max_buff = mh . size - cur_cnt ; int n = read ( sc -> serving_fd , marker , max_buff ) ; CHECK ( n >= 0 ) << " " ; marker = marker + n ; cur_cnt = cur_cnt + n ; } QueuedMessage * mq = new QueuedMessage ( mh . type , mh . size , read_buffer ) ; if ( mh . type == DIFF ) sc -> receive_queue . push ( mq ) ; else sc -> receive_queue_ctrl . push ( mq ) ; } return NULL ; }
Handle each peer connection in their own handlers
Handle each peer connection in their own handlers
// FIXME: The condition where socket server gets a connection // from the peer but the user didn't allocate a SocketChannel // for the peer.This can even happen if adapter is instantiated // first and SocketChannel is allocated for the peer subsequently // For now assume the user will maintain the right order i.e // allocate a SocketChannel for all peers and then instantiate // the adapter. /*      if(cd->sa->channels.size() < mh.rank || cd->sq->channels.at(mh.rank) == NULL){
            printf("ERROR:No SocketChannel assigned for the peer with rank [%d]...terminating the thread handler\n", mh.rank);
            // FIXME: Notify the client and exit
            pthread_exit();
            }
    */ // From the peer rank locate the local SocketChannel for that // peer and sets it's serving_fd // Wrap the received message in an object QueuedMessage and // push it to the local receive queue of the peer's // SocketChannel