WebSVN – HelenOS – Diff – //branches/network/uspace/srv/net/socket/socket_client.c

  *  @see socket
  */
 typedef socket_t *  socket_ref;
 /** Socket specific data.
+ *  Each socket lock locks only its structure part and any number of them may be locked simultaneously.
  */
 struct socket{
     /** Socket identifier.
      */
     int                 socket_id;
     /** Underlying protocol header size.
      *  Sending and receiving optimalization.
      */
     size_t              header_size;
     /** Packet data fragment size.
-     *  Sending and receiving optimalization.
+     *  Sending optimalization.
      */
     size_t              data_fragment_size;
+    /** Sending safety lock.
+     *  Locks the header_size and data_fragment_size attributes.
+     */
+    fibril_rwlock_t     sending_lock;
     /** Received packets queue.
      */
     dyn_fifo_t          received;
     /** Received packets safety lock.
+     *  Used for receiving and receive notifications.
+     *  Locks the received attribute.
      */
     fibril_mutex_t      receive_lock;
     /** Received packets signaling.
+     *  Signaled upon receive notification.
      */
     fibril_condvar_t    receive_signal;
     /** Waiting sockets queue.
      */
     dyn_fifo_t          accepted;
     /** Waiting sockets safety lock.
+     *  Used for accepting and accept notifications.
+     *  Locks the accepted attribute.
      */
     fibril_mutex_t      accept_lock;
     /** Waiting sockets signaling.
+     *  Signaled upon accept notification.
      */
     fibril_condvar_t    accept_signal;
+    /** The number of blocked functions called.
+     *  Used while waiting for the received packets or accepted sockets.
+     */
+    int                 blocked;
 };
 /** Sockets map.
  *  Maps socket identifiers to the socket specific data.
  *  @see int_map.h
  */
 INT_MAP_DECLARE( sockets, socket_t );
 /** Socket client library global data.
  */
-static struct{
+static struct socket_client_globals {
     /** TCP module phone.
      */
     int tcp_phone;
     /** UDP module phone.
      */
     int udp_phone;
     /** Active sockets.
      */
     sockets_ref sockets;
+    /** Safety lock.
+     *  Write lock is used only for adding or removing sockets.
+     *  When locked for writing, no other socket locks need to be locked.
+     *  When locked for reading, any other socket locks may be locked.
+     *  No socket lock may be locked if this lock is unlocked.
+     */
+    fibril_rwlock_t lock;
-} socket_globals = { -1, -1, NULL };
+} socket_globals = {
+    .tcp_phone = -1,
+    .udp_phone = -1,
+    .sockets = NULL,
+    .lock = {
+        .readers = 0,
+        .writers = 0,
+        .waiters = {
+            .prev = & socket_globals.lock.waiters,
+            .next = & socket_globals.lock.waiters
+        }
+    }
+};
 INT_MAP_IMPLEMENT( sockets, socket_t );
 /** Returns the TCP module phone.
  *  Connects to the TCP module if necessary.
     dyn_fifo_initialize( & socket->accepted, SOCKET_INITIAL_ACCEPTED_SIZE );
     fibril_mutex_initialize( & socket->receive_lock );
     fibril_condvar_initialize( & socket->receive_signal );
     fibril_mutex_initialize( & socket->accept_lock );
     fibril_condvar_initialize( & socket->accept_signal );
+    fibril_rwlock_initialize( & socket->sending_lock );
+}
 void socket_connection( ipc_callid_t iid, ipc_call_t * icall ){
     ERROR_DECLARE;
     while( true ){
         callid = async_get_call( & call );
         switch( IPC_GET_METHOD( call )){
+                // TODO remember the data_fragment_size
             case NET_SOCKET_RECEIVED:
+                fibril_rwlock_read_lock( & socket_globals.lock );
                 // find the socket
                 socket = sockets_find( socket_get_sockets(), SOCKET_GET_SOCKET_ID( call ));
                 if( ! socket ){
                     ERROR_CODE = ENOTSOCK;
-                    break;
-                }
+                }else{
-                fibril_mutex_lock( & socket->receive_lock );
+                    fibril_mutex_lock( & socket->receive_lock );
-                // push the number of received packet fragments
+                    // push the number of received packet fragments
-                if( ! ERROR_OCCURRED( dyn_fifo_push( & socket->received, SOCKET_GET_DATA_FRAGMENTS( call ), SOCKET_MAX_RECEIVED_SIZE ))){
+                    if( ! ERROR_OCCURRED( dyn_fifo_push( & socket->received, SOCKET_GET_DATA_FRAGMENTS( call ), SOCKET_MAX_RECEIVED_SIZE ))){
-                    // signal the received packet
+                        // signal the received packet
-                    fibril_condvar_signal( & socket->receive_signal );
+                        fibril_condvar_signal( & socket->receive_signal );
+                    }
+                    fibril_mutex_unlock( & socket->receive_lock );
+                }
-                fibril_mutex_unlock( & socket->receive_lock );
+                fibril_rwlock_read_unlock( & socket_globals.lock );
                 break;
             case NET_SOCKET_ACCEPTED:
+                fibril_rwlock_read_lock( & socket_globals.lock );
                 // find the socket
                 socket = sockets_find( socket_get_sockets(), SOCKET_GET_SOCKET_ID( call ));
                 if( ! socket ){
                     ERROR_CODE = ENOTSOCK;
-                    break;
-                }
-                // create a new scoket
-                new_socket = ( socket_ref ) malloc( sizeof( socket_t ));
-                if( ! new_socket ){
-                    ERROR_CODE = ENOMEM;
-                    break;
-                }
-                socket_initialize( new_socket, SOCKET_GET_SOCKET_ID( call ), socket->phone, socket->service );
-                ERROR_CODE = sockets_add( socket_get_sockets(), new_socket->socket_id, new_socket );
-                if( ERROR_CODE < 0 ){
-                    free( new_socket );
                 }else{
-                    // push the new socket identifier
+                    // create a new scoket
-                    fibril_mutex_lock( & socket->accept_lock );
-                    if( ERROR_OCCURRED( dyn_fifo_push( & socket->accepted, new_socket->socket_id, SOCKET_MAX_ACCEPTED_SIZE ))){
-                        sockets_exclude( socket_get_sockets(), new_socket->socket_id );
+                    new_socket = ( socket_ref ) malloc( sizeof( socket_t ));
-                        free( new_socket );
+                    if( ! new_socket ){
+                        ERROR_CODE = ENOMEM;
                     }else{
+                        socket_initialize( new_socket, SOCKET_GET_SOCKET_ID( call ), socket->phone, socket->service );
+                        ERROR_CODE = sockets_add( socket_get_sockets(), new_socket->socket_id, new_socket );
+                        if( ERROR_CODE < 0 ){
+                            free( new_socket );
+                        }else{
+                            // push the new socket identifier
+                            fibril_mutex_lock( & socket->accept_lock );
+                            if( ERROR_OCCURRED( dyn_fifo_push( & socket->accepted, new_socket->socket_id, SOCKET_MAX_ACCEPTED_SIZE ))){
+                                sockets_exclude( socket_get_sockets(), new_socket->socket_id );
+                                free( new_socket );
+                            }else{
-                        // signal the accepted socket
+                                // signal the accepted socket
-                        fibril_condvar_signal( & socket->accept_signal );
+                                fibril_condvar_signal( & socket->accept_signal );
+                            }
+                            fibril_mutex_unlock( & socket->accept_lock );
+                            ERROR_CODE = EOK;
+                        }
+                    }
-                    fibril_mutex_unlock( & socket->accept_lock );
-                    ERROR_CODE = EOK;
+                }
+                fibril_rwlock_read_unlock( & socket_globals.lock );
                 break;
+            // TODO obsolete?
             case NET_SOCKET_DATA_FRAGMENT_SIZE:
+                fibril_rwlock_read_lock( & socket_globals.lock );
                 // find the socket
                 socket = sockets_find( socket_get_sockets(), SOCKET_GET_SOCKET_ID( call ));
                 if( ! socket ){
                     ERROR_CODE = ENOTSOCK;
-                    break;
+                }else{
+                    fibril_rwlock_write_lock( & socket->sending_lock );
+                    // set the data fragment size
+                    socket->data_fragment_size = SOCKET_GET_DATA_FRAGMENT_SIZE( call );
+                    fibril_rwlock_write_unlock( & socket->sending_lock );
+                    ERROR_CODE = EOK;
+                }
-                // set the data fragment size
-                socket->data_fragment_size = SOCKET_GET_DATA_FRAGMENT_SIZE( call );
+                fibril_rwlock_read_unlock( & socket_globals.lock );
-                ERROR_CODE = EOK;
                 break;
             default:
                 ERROR_CODE = ENOTSUP;
+        }
         ipc_answer_0( callid, ( ipcarg_t ) ERROR_CODE );
         dyn_fifo_destroy( & socket->received );
         free( socket );
         return ERROR_CODE;
+    }
     // request a new socket
-    if( ERROR_OCCURRED(( int ) async_req_3_3( phone, NET_SOCKET, 0, 0, service, ( ipcarg_t * ) & socket_id, ( ipcarg_t * ) & socket->header_size, ( ipcarg_t * ) & socket->data_fragment_size ))){
+    if( ERROR_OCCURRED(( int ) async_req_3_3( phone, NET_SOCKET, 0, 0, service, ( ipcarg_t * ) & socket_id, ( ipcarg_t * ) & socket->data_fragment_size, ( ipcarg_t * ) & socket->header_size ))){
         dyn_fifo_destroy( & socket->received );
         dyn_fifo_destroy( & socket->accepted );
         free( socket );
         return ERROR_CODE;
+    }
     // finish the new socket initialization
     socket_initialize( socket, socket_id, phone, service );
     // store the new socket
+    fibril_rwlock_write_lock( & socket_globals.lock );
     ERROR_CODE = sockets_add( socket_get_sockets(), socket_id, socket );
+    fibril_rwlock_write_unlock( & socket_globals.lock );
     if( ERROR_CODE < 0 ){
         dyn_fifo_destroy( & socket->received );
         dyn_fifo_destroy( & socket->accepted );
         free( socket );
         async_msg_3( phone, NET_SOCKET_CLOSE, ( ipcarg_t ) socket_id, 0, service );
     aid_t           message_id;
     ipcarg_t        result;
     if( ! data ) return EBADMEM;
     if( ! datalength ) return NO_DATA;
+    fibril_rwlock_read_lock( & socket_globals.lock );
     // find the socket
     socket = sockets_find( socket_get_sockets(), socket_id );
+    if( ! socket ){
+        fibril_rwlock_read_unlock( & socket_globals.lock );
-    if( ! socket ) return ENOTSOCK;
+        return ENOTSOCK;
+    }
     // request the message
     message_id = async_send_3( socket->phone, message, ( ipcarg_t ) socket->socket_id, arg2, socket->service, NULL );
     // send the address
     ipc_data_write_start( socket->phone, data, datalength );
+    fibril_rwlock_read_unlock( & socket_globals.lock );
     async_wait_for( message_id, & result );
     return ( int ) result;
+}
 int bind( int socket_id, const struct sockaddr * my_addr, socklen_t addrlen ){
     // send the address
     return socket_send_data( socket_id, NET_SOCKET_BIND, 0, my_addr, ( size_t ) addrlen );
+}
 int listen( int socket_id, int backlog ){
-    socket_ref      socket;
+    socket_ref  socket;
+    int         result;
     if( backlog <= 0 ) return EINVAL;
+    fibril_rwlock_read_lock( & socket_globals.lock );
     // find the socket
     socket = sockets_find( socket_get_sockets(), socket_id );
+    if( ! socket ){
+        fibril_rwlock_read_unlock( & socket_globals.lock );
-    if( ! socket ) return ENOTSOCK;
+        return ENOTSOCK;
+    }
     // request listen backlog change
-    return ( int ) async_req_3_0( socket->phone, NET_SOCKET_LISTEN, ( ipcarg_t ) socket->socket_id, ( ipcarg_t ) backlog, socket->service );
+    result = ( int ) async_req_3_0( socket->phone, NET_SOCKET_LISTEN, ( ipcarg_t ) socket->socket_id, ( ipcarg_t ) backlog, socket->service );
+    fibril_rwlock_read_unlock( & socket_globals.lock );
+    return result;
+}
 int accept( int socket_id, struct sockaddr * cliaddr, socklen_t * addrlen ){
     socket_ref      socket;
     aid_t           message_id;
     int             result;
     if(( ! cliaddr ) || ( ! addrlen )) return EBADMEM;
+    fibril_rwlock_read_lock( & socket_globals.lock );
     // find the socket
     socket = sockets_find( socket_get_sockets(), socket_id );
+    if( ! socket ){
+        fibril_rwlock_read_unlock( & socket_globals.lock );
-    if( ! socket ) return ENOTSOCK;
+        return ENOTSOCK;
+    }
     fibril_mutex_lock( & socket->accept_lock );
     // wait for an accepted socket
+    ++ socket->blocked;
     while( dyn_fifo_value( & socket->accepted ) <= 0 ){
+        fibril_rwlock_read_unlock( & socket_globals.lock );
         fibril_condvar_wait( & socket->accept_signal, & socket->accept_lock );
+        fibril_rwlock_read_lock( & socket_globals.lock );
+    }
+    -- socket->blocked;
     // request accept
     message_id = async_send_3( socket->phone, NET_SOCKET_ACCEPT, ( ipcarg_t ) socket->socket_id, ( ipcarg_t ) dyn_fifo_value( & socket->accepted ), socket->service, NULL );
     // read address
     ipc_data_read_start( socket->phone, cliaddr, * addrlen );
+    fibril_rwlock_read_unlock( & socket_globals.lock );
     async_wait_for( message_id, ( ipcarg_t * ) & result );
     if( result > 0 ){
         // dequeue the accepted apcket if successful
         dyn_fifo_pop( & socket->accepted );
+    }
 int closesocket( int socket_id ){
     ERROR_DECLARE;
     socket_ref      socket;
+    fibril_rwlock_write_unlock( & socket_globals.lock );
     socket = sockets_find( socket_get_sockets(), socket_id );
+    if( ! socket ){
+        fibril_rwlock_write_unlock( & socket_globals.lock );
-    if( ! socket ) return ENOTSOCK;
+        return ENOTSOCK;
+    }
+    if( socket->blocked ){
+        fibril_rwlock_write_unlock( & socket_globals.lock );
+        return EINPROGRESS;
+    }
     // request close
     ERROR_PROPAGATE(( int ) async_req_3_0( socket->phone, NET_SOCKET_CLOSE, ( ipcarg_t ) socket->socket_id, 0, socket->service ));
     // free the socket structure
     socket_destroy( socket );
+    fibril_rwlock_write_unlock( & socket_globals.lock );
     return EOK;
+}
 void socket_destroy( socket_ref socket ){
     int accepted_id;
     ipcarg_t        result;
     size_t          fragments;
     if( ! data ) return EBADMEM;
     if( ! datalength ) return NO_DATA;
+    fibril_rwlock_read_lock( & socket_globals.lock );
     // find socket
     socket = sockets_find( socket_get_sockets(), socket_id );
+    if( ! socket ){
+        fibril_rwlock_read_unlock( & socket_globals.lock );
-    if( ! socket ) return ENOTSOCK;
+        return ENOTSOCK;
+    }
+    fibril_rwlock_read_lock( & socket->sending_lock );
     // compute data fragment count
     fragments = ( datalength + socket->header_size ) / socket->data_fragment_size;
     if(( datalength + socket->header_size ) % socket->data_fragment_size ) ++ fragments;
     // request send
-    message_id = async_send_4( socket->phone, message, ( ipcarg_t ) socket->socket_id, fragments, socket->service, ( ipcarg_t ) flags, NULL );
+    message_id = async_send_5( socket->phone, message, ( ipcarg_t ) socket->socket_id, socket->data_fragment_size, socket->service, ( ipcarg_t ) flags, fragments, NULL );
     // send the address if given
     if(( ! toaddr ) || ( ipc_data_write_start( socket->phone, toaddr, addrlen ) == EOK )){
         if( fragments == 1 ){
             // send all if only one fragment
             ipc_data_write_start( socket->phone, data, datalength );
+            }
             // send the last fragment
             ipc_data_write_start( socket->phone, data, ( datalength + socket->header_size ) % socket->data_fragment_size );
+        }
+    }
+    fibril_rwlock_read_unlock( & socket->sending_lock );
+    fibril_rwlock_read_unlock( & socket_globals.lock );
     async_wait_for( message_id, & result );
     return ( int ) result;
+}
 int recv( int socket_id, void * data, size_t datalength, int flags ){
     ipc_call_t      answer;
     if( ! data ) return EBADMEM;
     if( ! datalength ) return NO_DATA;
     if( fromaddr && ( ! addrlen )) return EINVAL;
+    fibril_rwlock_read_lock( & socket_globals.lock );
     // find the socket
     socket = sockets_find( socket_get_sockets(), socket_id );
+    if( ! socket ){
+        fibril_rwlock_read_unlock( & socket_globals.lock );
-    if( ! socket ) return ENOTSOCK;
+        return ENOTSOCK;
+    }
     fibril_mutex_lock( & socket->receive_lock );
     // wait for a received packet
+    ++ socket->blocked;
     while(( result = dyn_fifo_value( & socket->received )) <= 0 ){
+        fibril_rwlock_read_unlock( & socket_globals.lock );
         fibril_condvar_wait( & socket->receive_signal, & socket->receive_lock );
+        fibril_rwlock_read_lock( & socket_globals.lock );
+    }
+    -- socket->blocked;
     fragments = ( size_t ) result;
     // prepare lengths if more fragments
     if( fragments > 1 ){
         lengths = ( size_t * ) malloc( sizeof( size_t ) * fragments + sizeof( size_t ));
         if( ! lengths ){
             fibril_mutex_unlock( & socket->receive_lock );
+            fibril_rwlock_read_unlock( & socket_globals.lock );
             return ENOMEM;
+        }
         // request packet data
         message_id = async_send_4( socket->phone, message, ( ipcarg_t ) socket->socket_id, 0, socket->service, ( ipcarg_t ) flags, & answer );
         // read the address if desired
         result = SOCKET_GET_READ_DATA_LENGTH( answer );
         // set address length
         if( fromaddr && addrlen ) * addrlen = SOCKET_GET_ADDRESS_LENGTH( answer );
+    }
     fibril_mutex_unlock( & socket->receive_lock );
+    fibril_rwlock_read_unlock( & socket_globals.lock );
     return result;
+}
 int getsockopt( int socket_id, int level, int optname, void * value, size_t * optlen ){
     socket_ref      socket;
     aid_t           message_id;
     ipcarg_t        result;
     if( !( value && optlen )) return EBADMEM;
     if( !( * optlen )) return NO_DATA;
+    fibril_rwlock_read_lock( & socket_globals.lock );
     // find the socket
     socket = sockets_find( socket_get_sockets(), socket_id );
+    if( ! socket ){
+        fibril_rwlock_read_unlock( & socket_globals.lock );
-    if( ! socket ) return ENOTSOCK;
+        return ENOTSOCK;
+    }
     // request option value
     message_id = async_send_3( socket->phone, NET_SOCKET_GETSOCKOPT, ( ipcarg_t ) socket->socket_id, ( ipcarg_t ) optname, socket->service, NULL );
     // read the length
     if( ipc_data_read_start( socket->phone, optlen, sizeof( * optlen )) == EOK ){
         // read the value
         ipc_data_read_start( socket->phone, value, * optlen );
+    }
+    fibril_rwlock_read_unlock( & socket_globals.lock );
     async_wait_for( message_id, & result );
     return ( int ) result;
+}
 int setsockopt( int socket_id, int level, int optname, const void * value, size_t optlen ){

Subversion Repositories HelenOS

(root)//branches/network/uspace/srv/net/socket/socket_client.c – Rev 4708 → 4726