Nilorea Library

- Main Page
- Modules
- Data Structures
- Files

- File List
- Globals

All Data Structures Files Functions Variables Typedefs Macros Groups
n_network.c
Go to the documentation of this file.
1
8 #include "n_common.h"
9 #include "n_log.h"
10 #include "n_str.h"
11 #include "n_list.h"
12 #include "n_network.h"
13
14
15
23 int handle_wsa( int mode , int v1 , int v2 )
24 {
25 int compiler_warning_suppressor = 0 ;
26 #if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX )
27 static WSADATA WSAdata; /*WSA world*/
28 static int WSA_IS_INITIALIZED = 0; /*status checking*/
29
30 switch ( mode )
31 {
32 /*returning WSA status*/
33 case 2 :
34 return WSA_IS_INITIALIZED;
35 break;
36 /*loading WSA dll*/
37 case 1 :
38 if ( WSA_IS_INITIALIZED == 1 )
39 return TRUE; /*already loaded*/
40 if ( ( WSAStartup( MAKEWORD( v1 , v2 ), &WSAdata ) ) != 0 )
41 {
42 WSA_IS_INITIALIZED = 0;
43 return FALSE;
44 }
45 else
46 {
47 WSA_IS_INITIALIZED = 1 ;
48 return TRUE;
49 }
50 break;
51 /*unloading (closing) WSA */
52 case 0 :
53 if ( WSA_IS_INITIALIZED == 0 )
54 return TRUE; /*already CLEANED or not loaded */
55 if ( WSACleanup() == 0 )
56 {
57 WSA_IS_INITIALIZED = 0;
58 return TRUE;
59 }
60 break;
61 } /*switch(...)*/
62 #endif /*#ifndef LINUX SOLARIS AIX */
63 compiler_warning_suppressor = mode + v1 + v2 ;
64 compiler_warning_suppressor = TRUE ;
65 return compiler_warning_suppressor ;
66 } /*handle_wsa(...)*/
67
68
69
78 int netw_setsockopt( SOCKET sock , int disable_naggle , int sock_send_buf , int sock_recv_buf )
79 {
80 /* disable naggle algorithm */
81 if( disable_naggle > 0 )
82 {
83 int tmp = 1;
84 if ( setsockopt( sock , IPPROTO_TCP, TCP_NODELAY, ( const char * ) &tmp, sizeof( tmp ) ) == -1 )
85 {
86 n_log( LOG_ERR , "Error from setsockopt(TCP_NODELAY). errno: %s" , strerror( errno ) );
87 return FALSE ;
88 }
89 }
90 /* socket sending buffer size */
91 if( sock_send_buf > 0 )
92 {
93 if ( setsockopt ( sock , SOL_SOCKET, SO_SNDBUF, ( const char * ) &sock_send_buf, sizeof( sock_send_buf ) ) == -1 )
94 {
95 n_log( LOG_ERR , "Error from setsockopt(SO_SNDBUF). errno: %s" , strerror( errno ) );
96 return FALSE ;
97 }
98 }
99 /* socket receiving buffer */
100 if( sock_recv_buf > 0 )
101 {
102 if ( setsockopt ( sock , SOL_SOCKET, SO_RCVBUF, ( const char * ) &sock_recv_buf, sizeof( sock_recv_buf ) ) == -1 )
103 {
104 n_log( LOG_ERR , "Error from setsockopt(SO_RCVBUF). errno: %s" , strerror( errno ) );
105 return FALSE ;
106 }
107 }
108 /* lose the pesky "Address already in use" error message*/
109 if ( setsockopt( sock , SOL_SOCKET, SO_REUSEADDR , "1", sizeof( int ) ) == -1 )
110 {
111 n_log( LOG_ERR , "Error from setsockopt(SO_REUSEADDR). errno: s" , strerror( errno ) );
112 return FALSE ;
113 }
114 return TRUE ;
115 } /* netw_setsockopt( ... ) */
116
117
118
124 int netw_tcp_socket( SOCKET *sock )
125 {
126 int tmpsock = 0 ;
127
128 /* new socket */
129 tmpsock = socket( AF_INET , SOCK_STREAM , 0 );
130 if( tmpsock < 0 )
131 {
132 n_log( LOG_ERR , "Error from socket(). errno: %s" , strerror( errno ) );
133 return FALSE ;
134 }
135 (*sock) = tmpsock ;
136
137 return TRUE ;
138 } /* new_tcp_socket( ... ) */
139
140
141
154 int netw_connect_ex( NETWORK **netw , char *host , int port, int disable_naggle , int sock_send_buf , int sock_recv_buf , int send_list_limit , int recv_list_limit )
155 {
156 struct hostent *hostinfo = NULL ;
157
158 /*checking WSA when under windows*/
159 if ( handle_wsa( 1 , 2 , 2 ) == FALSE )
160 {
161 n_log( LOG_ERR , "Unable to load WSA dll's" );
162 return FALSE ;
163 }
164
165 /*do not work over an already used netw*/
166 if( (*netw) )
167 {
168 n_log( LOG_ERR , "Unable to allocate (*netw), already existing. You must use empty NETWORK *structs." );
169 return FALSE ;
170 }
171
172 hostinfo = gethostbyname( host );
173 if( hostinfo == NULL )
174 {
175 n_log( LOG_ERR , "Erreur -> host : %s inconnu.", host );
176 return FALSE ;
177 }
178
179 /*creating array*/
180 Malloc( ( *netw ) , NETWORK , 1 );
181 _assert( (*netw) , return FALSE );
182
183 /* new socket */
184 if( netw_tcp_socket( &(*netw) -> link . sock ) == FALSE )
185 {
186 n_log( LOG_ERR , "Error from socket, %s" , strerror( errno ) );
187 Free( (*netw ) );
188 return FALSE ;
189 }
190
191 if( netw_setsockopt( (*netw) -> link . sock , disable_naggle , sock_send_buf , sock_recv_buf ) != TRUE )
192 {
193 n_log( LOG_ERR , "Some socket options could not be set on %d" , (*netw) -> link . sock );
194 }
195
196 /*filling host information*/
197 memcpy(&( *netw ) -> link . raddr . sin_addr , hostinfo -> h_addr_list[ 0 ] , hostinfo -> h_length );
198 ( *netw ) -> link . raddr . sin_family = AF_INET;
199 ( *netw ) -> link . raddr . sin_port = htons( port );
200
201 /*filling local information*/
202 ( *netw ) -> link . haddr . sin_addr . s_addr = htonl( INADDR_ANY );
203 ( *netw ) -> link . haddr . sin_family = AF_INET;
204 ( *netw ) -> link . haddr . sin_port = htons( port );
205 ( *netw ) -> link . port = port;
206
207 /*zero the rest of the struct */
208 memset( &( ( *netw ) -> link . haddr . sin_zero ), 0 , 8 );
209 memset( &( ( *netw ) -> link . raddr . sin_zero ), 0 , 8 );
210
211 /*connecting*/
212 if( connect( ( *netw ) -> link .sock , ( struct sockaddr * ) &( ( *netw ) -> link . raddr ) , sizeof( struct sockaddr ) ) != 0 )
213 {
214 n_log( LOG_ERR , "Error from connect(). errno: %s" , strerror( errno ) );
215 closesocket( (*netw) -> link . sock );
216 Free( (*netw) );
217 return FALSE ;
218 }
219
220 /*storing connected to ip adress*/
221 ( *netw ) -> link . ip = strdup( host );
222
223 /*initialize queues */
224 ( *netw ) -> recv_buf = new_generic_list( recv_list_limit );
225 if( ! (*netw ) -> recv_buf )
226 {
227 n_log( LOG_ERR , "Error when creating receive list with %d item limit" , recv_list_limit );
228 closesocket( (*netw) -> link . sock );
229 Free( (*netw) );
230 return FALSE ;
231 }
232 (*netw) -> send_buf = new_generic_list( send_list_limit );
233 if( ! (*netw ) -> send_buf )
234 {
235 n_log( LOG_ERR , "Error when creating send list with %d item limit" , send_list_limit );
236 list_destroy( &( *netw ) -> recv_buf );
237 closesocket( (*netw) -> link . sock );
238 Free( (*netw) );
239 return FALSE ;
240 }
241
242 /*initiliaze mutexs*/
243 if ( pthread_mutex_init( &( *netw ) -> sendbolt , NULL ) != 0 )
244 {
245 n_log( LOG_ERR , "Error initializing (*netw) -> sendbolt" );
246 list_destroy( &( *netw ) -> recv_buf );
247 list_destroy( &( *netw ) -> send_buf );
248 closesocket( (*netw) -> link . sock );
249 Free( (*netw) );
250 return FALSE ;
251 }
252 if ( pthread_mutex_init( &( *netw ) -> recvbolt , NULL ) != 0 )
253 {
254 n_log( LOG_ERR , "Error initializing (*netw) -> recvbolt" );
255 pthread_mutex_destroy( &( *netw ) -> sendbolt );
256 list_destroy( &( *netw ) -> recv_buf );
257 list_destroy( &( *netw ) -> send_buf );
258 closesocket( (*netw) -> link . sock );
259 Free( (*netw) );
260 return FALSE ;
261 }
262 if ( pthread_mutex_init( &( *netw ) -> eventbolt , NULL ) != 0 )
263 {
264 n_log( LOG_ERR , "Error initializing (*netw) -> eventbolt" );
265 pthread_mutex_destroy( &( *netw ) -> sendbolt );
266 pthread_mutex_destroy( &( *netw ) -> recvbolt );
267 list_destroy( &( *netw ) -> recv_buf );
268 list_destroy( &( *netw ) -> send_buf );
269 closesocket( (*netw) -> link . sock );
270 Free( (*netw) );
271 return FALSE ;
272 }
273 (*netw) -> state = NETW_RUN ;
274 (*netw) -> threaded_engine_status = NETW_THR_ENGINE_STOPPED ;
275
276 return TRUE ;
277 } /* netw_connect_ex(...)*/
278
279
280
288 int netw_connect( NETWORK **netw , char *host , int port )
289 {
290 return netw_connect_ex( &(*netw) , host , port , 0 , 0 , 0 , 0 , 0 );
291 } /* netw_connect() */
292
293
294
301 int netw_set( NETWORK *netw , int flag )
302 {
303 /* empty send queue */
304 if( flag & NETW_EMPTY_SENDBUF )
305 {
306 pthread_mutex_lock( &netw -> sendbolt );
307 list_empty( netw -> send_buf );
308 pthread_mutex_unlock( &netw -> sendbolt );
309 };
310 /* empty recv queue */
311 if( flag & NETW_EMPTY_RECVBUF )
312 {
313 pthread_mutex_lock( &netw -> recvbolt );
314 list_empty( netw -> recv_buf );
315 pthread_mutex_unlock( &netw -> recvbolt );
316 }
317 /* pause network */
318 if( flag & NETW_PAUSE )
319 {
320 pthread_mutex_lock( &netw -> eventbolt );
321 netw -> state = NETW_PAUSE ;
322 pthread_mutex_unlock( &netw -> eventbolt );
323 }
324 /* unpause network */
325 if( flag & NETW_RUN )
326 {
327 pthread_mutex_lock( &netw -> eventbolt );
328 netw -> state = NETW_RUN ;
329 pthread_mutex_unlock( &netw -> eventbolt );
330 }
331 return TRUE;
332 } /* netw_set(...) */
333
334
335
341 int netw_close( NETWORK **netw )
342 {
343 _assert( (*netw) , return FALSE );
344
345 /*closing connection*/
346 closesocket( (*netw) -> link . sock );
347
348 if( (*netw) -> link . ip )
349 {
350 Free( (*netw) -> link . ip );
351 }
352
353 /*list freeing*/
354 list_destroy( &(*netw) -> send_buf );
355 list_destroy( &(*netw) -> recv_buf );
356
357
358 pthread_mutex_destroy( &(*netw) -> recvbolt );
359 pthread_mutex_destroy( &(*netw) -> sendbolt );
360 pthread_mutex_destroy( &(*netw) -> eventbolt );
361
362 Free( (*netw) );
363
364 return TRUE;
365 } /* netw_close(...)*/
366
367
368
376 int netw_make_listening( NETWORK **netw , int PORT , int nbpending )
377 {
378 /*checking WSA when under windows*/
379 if ( handle_wsa( 1 , 2 , 2 ) == FALSE )
380 {
381 n_log( LOG_ERR , "Unable to load WSA dll's" );
382 return FALSE ;
383 }
384
385 if( *netw )
386 {
387 n_log( LOG_ERR , "Cannot use an allocated network. Please pass a NULL network to modify" );
388 return FALSE;
389 }
390
391 /*creating array*/
392 Malloc( *netw , NETWORK , 1 );
393
394 if( netw_tcp_socket( &(*netw) -> link . sock ) == FALSE )
395 {
396 n_log( LOG_ERR , "Error from socket, %s" , strerror( errno ) );
397 Free( (*netw ) );
398 return FALSE ;
399 }
400
401 if( netw_setsockopt( (*netw) -> link . sock , 0 , 0 , 0 ) != TRUE )
402 {
403 n_log( LOG_ERR , "Some socket options could not be set on %d" , (*netw) -> link . sock );
404 }
405
406 /*filling local information*/
407 ( *netw ) -> link . haddr . sin_addr.s_addr = INADDR_ANY;
408 ( *netw ) -> link . haddr . sin_family = AF_INET;
409 ( *netw ) -> link . haddr . sin_port = htons( PORT );
410 ( *netw ) -> nb_pending = nbpending;
411
412 memset( &( ( *netw ) -> link . haddr . sin_zero ), 0 , 8 ); /* zero the rest of the struct*/
413 memset( &( ( *netw ) -> link . raddr . sin_zero ), 0 , 8 ); /* zero the rest of the struct*/
414
415 /*binding information to socket*/
416 if ( bind( ( *netw ) -> link . sock , ( struct sockaddr * ) &( ( *netw ) -> link . haddr ) , sizeof( struct sockaddr ) ) == -1 )
417 {
418 n_log( LOG_ERR , "Error binding socket %d" , ( *netw ) -> link . sock );
419 return FALSE;
420 }
421
422 /* nb_pending connections*/
423 listen( ( *netw ) -> link . sock, ( *netw ) -> nb_pending );
424
425 (*netw) -> mode = NETW_SERVER ;
426 (*netw) -> state = NETW_RUN ;
427 (*netw) -> threaded_engine_status = NETW_THR_ENGINE_STOPPED ;
428
429 return TRUE;
430 } /* netw_make_listening(...)*/
431
432
433
444 NETWORK *netw_accept_from_ex( NETWORK *from , int disable_naggle , int sock_send_buf , int sock_recv_buf , int send_list_limit , int recv_list_limit )
445 {
446 int tmp = 0 ;
447
448 #if defined( LINUX ) || defined( SOLARIS ) || defined( AIX )
449 socklen_t sin_size = 0 ;
450 #else
451 int sin_size = 0 ;
452 #endif
453
454 NETWORK *netw = NULL ;
455
456 /*checking WSA when under windows*/
457 if( handle_wsa( 1 , 2 , 2 ) == FALSE )
458 {
459 n_log( LOG_ERR , "Unable to load WSA dll's" );
460 return FALSE ;
461 }
462
463 _assert( from , return FALSE );
464
465 Malloc( netw , NETWORK , 1 );
466
467 sin_size = sizeof( struct sockaddr_in );
468
469 tmp = accept( from -> link . sock, ( struct sockaddr * ) & ( netw -> link . raddr ) , &sin_size );
470 if ( tmp < 0 )
471 {
472 n_log( LOG_ERR , "error accepting on %d" , netw -> link . sock );
473 Free( netw );
474 return NULL;
475 }
476
477 netw -> link . sock = tmp ;
478 netw -> link . port = from -> link . port;
479
480 if( netw_setsockopt( netw -> link . sock , disable_naggle , sock_send_buf , sock_recv_buf ) != TRUE )
481 {
482 n_log( LOG_ERR , "Some socket options could not be set on %d" , netw-> link . sock );
483 }
484 /*initialize queues */
485 netw -> recv_buf = new_generic_list( recv_list_limit );
486 if( !netw -> recv_buf )
487 {
488 n_log( LOG_ERR , "Error when creating receive list with %d item limit" , recv_list_limit );
489 closesocket( netw -> link . sock );
490 Free( netw );
491 return FALSE ;
492 }
493 netw -> send_buf = new_generic_list( send_list_limit );
494 if( ! netw -> send_buf )
495 {
496 n_log( LOG_ERR , "Error when creating send list with %d item limit" , send_list_limit );
497 list_destroy( &netw -> recv_buf );
498 closesocket( netw -> link . sock );
499 Free( netw );
500 return FALSE ;
501 }
502
503 /*initiliaze mutexs*/
504 if( pthread_mutex_init( &netw -> sendbolt , NULL ) != 0 )
505 {
506 n_log( LOG_ERR , "Error initializing netw -> sendbolt" );
507 list_destroy( &netw -> recv_buf );
508 list_destroy( &netw -> send_buf );
509 closesocket( netw -> link . sock );
510 Free( netw );
511 return FALSE ;
512 }
513 if( pthread_mutex_init( &netw -> recvbolt , NULL ) != 0 )
514 {
515 n_log( LOG_ERR , "Error initializing netw -> recvbolt" );
516 pthread_mutex_destroy( &netw -> sendbolt );
517 list_destroy( &netw -> recv_buf );
518 list_destroy( &netw -> send_buf );
519 closesocket( netw -> link . sock );
520 Free( netw );
521 return FALSE ;
522 }
523 if ( pthread_mutex_init( &netw -> eventbolt , NULL ) != 0 )
524 {
525 n_log( LOG_ERR , "Error initializing (*netw) -> eventbolt" );
526 pthread_mutex_destroy( &netw -> sendbolt );
527 pthread_mutex_destroy( &netw -> recvbolt );
528 list_destroy( &netw -> recv_buf );
529 list_destroy( &netw -> send_buf );
530 closesocket( netw -> link . sock );
531 Free( netw );
532 return FALSE ;
533 }
534
535 netw -> mode = NETW_SERVER ;
536 netw -> state = NETW_RUN ;
537 netw -> threaded_engine_status = NETW_THR_ENGINE_STOPPED ;
538 netw -> link . ip = NULL ;
539
540 return netw;
541 } /* netw_accept_from_ex(...) */
542
543
544
550 NETWORK *netw_accept_from( NETWORK *from )
551 {
552 return netw_accept_from_ex( from , 0 , 0 , 0 , 0 , 0 );
553 } /* network_accept_from( ... ) */
554
555
556
563 int netw_add_msg( NETWORK *netw , N_STR *msg )
564 {
565 _assert( netw , return FALSE );
566 _assert( msg , return FALSE );
567 _assert( msg -> data , return FALSE );
568
569 if( msg -> length <= 0 )
570 {
571 n_log( LOG_ERR , "Empty messages are not supported. msg(%p)->lenght=%d" , msg , msg -> length );
572 return FALSE;
573 }
574
575 pthread_mutex_lock( &netw -> sendbolt );
576
577 if( list_push( netw -> send_buf , msg , free_nstr_ptr ) == FALSE )
578 {
579 pthread_mutex_unlock( &netw -> sendbolt );
580 return FALSE;
581 }
582
583 pthread_mutex_unlock( &netw -> sendbolt );
584
585 return TRUE;
586 } /* netw_add_msg(...) */
587
588
589
595 N_STR *netw_get_msg( NETWORK *netw )
596 {
597 N_STR * ptr = NULL ;
598
599 _assert( netw , return NULL );
600
601 pthread_mutex_lock( &netw -> recvbolt );
602
603 ptr = list_shift( netw -> recv_buf , N_STR );
604
605 pthread_mutex_unlock( &netw -> recvbolt );
606
607 return ptr ;
608 } /* netw_get_msg(...)*/
609
610
611
619 N_STR *netw_wait_msg( NETWORK *netw , int refresh , int timeout )
620 {
621 N_STR *nstrptr = NULL ;
622 int DONE = 0 , timed = 0 ;
623
624 _assert( netw , return NULL );
625
626 if( timeout > 0 )
627 timed = 1 ;
628
629 while( !DONE )
630 {
631 nstrptr = netw_get_msg( netw );
632 if( nstrptr )
633 return nstrptr ;
634
635 usleep( refresh );
636 if( timed == 1 )
637 {
638 timeout -= refresh ;
639 if( timeout < 0 )
640 break ;
641 }
642 }
643 return NULL ;
644 } /* netw_wait_msg(...) */
645
646
647
653 int netw_start_thr_engine( NETWORK *netw )
654 {
655 _assert( netw , return FALSE );
656
657 pthread_mutex_lock( &netw -> eventbolt );
658 if( netw -> threaded_engine_status == NETW_THR_ENGINE_STARTED )
659 {
660 n_log( LOG_ERR , "THR Engine already started for network %p" , netw );
661 pthread_mutex_unlock( &netw -> eventbolt );
662 return FALSE ;
663 }
664
665 if( pthread_create( &netw -> recv_thr , NULL , netw_recv_func , (void *) netw ) != 0 )
666 {
667 n_log( LOG_ERR , "Unable to create recv_thread for network %p" , netw );
668 pthread_mutex_unlock( &netw -> eventbolt );
669 return FALSE ;
670 }
671 if( pthread_create( &netw -> send_thr , NULL , netw_send_func , (void *) netw ) != 0 )
672 {
673 n_log( LOG_ERR , "Unable to create send_thread for network %p" , netw );
674 pthread_mutex_unlock( &netw -> eventbolt );
675 return FALSE ;
676 }
677
678 netw -> threaded_engine_status = NETW_THR_ENGINE_STARTED ;
679
680 pthread_mutex_unlock( &netw -> eventbolt );
681
682 return TRUE ;
683 } /* netw_create_recv_thread(....) */
684
685
686
691 void *netw_send_func( void *NET )
692 {
693 int DONE = 0 ,
694 state = 0 ;
695
696 NSTRBYTE nboctet = 0 ;
697
698 char nboct[4];
699
700 N_STR *ptr = NULL ;
701
702 NETWORK *netw = (NETWORK *) NET ;
703
704 _assert( netw , return NULL );
705
706 do
707 {
708 pthread_mutex_lock( &netw -> eventbolt );
709 state = netw -> state ;
710 pthread_mutex_unlock( &netw -> eventbolt );
711
712 if( state == NETW_QUIT )
713 {
714 DONE = 3;
715
716 /* preparing the number of octet */
717 nboctet = NETW_QUIT ;
718 nboctet = htonl( nboctet );
719
720 memcpy( nboct , &nboctet , sizeof( NSTRBYTE ) );
721 n_log( LOG_DEBUG , "%d Sending Quit !" , netw -> link . sock );
722
723 /* sending the number of octet to receive on next message */
724 if ( send_data( netw -> link . sock , nboct , sizeof( NSTRBYTE ) ) == -1 )
725 DONE = 4 ;
726 }
727 else
728 {
729 if( state != NETW_PAUSE )
730 {
731 pthread_mutex_lock( &netw -> sendbolt );
732
733 ptr = list_shift( netw -> send_buf , N_STR ) ;
734 if( ptr && ptr -> length > 0 && ptr -> data )
735 {
736 n_log( LOG_DEBUG , "Sending ptr size %ld written %ld" , ptr -> length , ptr -> written );
737
738 /* preparing the number of octet */
739 nboctet = ptr -> length;
740 nboctet = htonl( nboctet );
741
742 memcpy( nboct , &nboctet , sizeof( NSTRBYTE ) );
743
744 /* sending the number of octet to receive on next message */
745 if ( !DONE && send_data( netw -> link . sock , nboct , sizeof( NSTRBYTE ) ) == -1 )
746 DONE = 1 ;
747
748 /* sending the data itself */
749 if ( !DONE && send_data( netw -> link . sock , ptr->data , ptr -> length ) == -1 )
750 DONE = 2 ;
751
752 free_nstr( &ptr );
753 pthread_mutex_unlock( &netw -> sendbolt );
754 #if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX )
755 rest( 0 );
756 #else
757 usleep( 0 ); /* we will have to make a sleep thing for that */
758 #endif
759 }
760 else
761 {
762 pthread_mutex_unlock( &netw -> sendbolt );
763 #if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX )
764 rest( 1 );
765 #else
766 usleep( 1000 );
767 #endif
768 }
769 }
770 else
771 {
772 #if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX )
773 rest( 10 );
774 #else
775 usleep( 10000 );
776 #endif
777 }
778 }
779 }while( !DONE );
780
781 if( DONE == 1 )
782 n_log( LOG_ERR , "Error when sending number of octet on socket %d" , netw -> link . sock );
783 else if( DONE == 2 )
784 n_log( LOG_ERR , "Error when sending data from socket %d" , netw -> link . sock );
785 else if( DONE == 3 )
786 n_log( LOG_DEBUG , "Quit asked, closing sending thread on socket %d" , netw -> link . sock );
787 else if( DONE == 4 )
788 n_log( LOG_ERR , "Error when sending number of QUIT on socket %d" , netw -> link . sock );
789 else
790 n_log( LOG_ERR , "Closing send thread for socket %d\nReason unknown\nDONE = %d" , netw -> link . sock , DONE );
791
792 pthread_mutex_lock( &netw -> eventbolt );
793 if( DONE == 4 )
794 netw -> state = NETW_QUIT ;
795 else
796 netw -> state = NETW_ERROR ;
797 pthread_mutex_unlock( &netw -> eventbolt );
798
799 n_log( LOG_DEBUG , "Socket %d: Sending thread exited !" , netw -> link . sock );
800
801 pthread_exit( 0 );
802
803 /* suppress compiler warning */
804 #if !defined( LINUX ) && !defined( SOLARIS )
805 return NULL ;
806 #endif
807 } /* netw_send_func(...) */
808
809
810
815 void *netw_recv_func( void *NET )
816 {
817 int DONE = 0 ,
818 state = 0 ,
819 tmpstate = 0 ;
820
821 NSTRBYTE nboctet = 0 ;
822
823 char nboct[4];
824
825 N_STR *recvdmsg = NULL ;
826
827 NETWORK *netw = (NETWORK *) NET ;
828
829 _assert( netw , return NULL );
830
831 do
832 {
833 pthread_mutex_lock( &netw -> eventbolt );
834 state = netw -> state ;
835 pthread_mutex_unlock( &netw -> eventbolt );
836
837 if( state == NETW_QUIT )
838 {
839 DONE = 4;
840 }
841 else
842 {
843 /* receiving the number of octet to receive on next message */
844 if( !DONE && state != NETW_PAUSE )
845 {
846 if( recv_data( netw -> link . sock , nboct , sizeof( NSTRBYTE ) ) == -1 )
847 DONE = 1 ;
848 else
849 {
850 memcpy( &nboctet , nboct , sizeof( NSTRBYTE ) );
851 tmpstate = ntohl( nboctet );
852 nboctet = tmpstate ;
853 if( tmpstate == NETW_QUIT )
854 {
855 n_log( LOG_DEBUG , "%d receiving order to QUIT, nboctet %d NETW_QUIT %d !" , netw -> link . sock , nboctet , NETW_QUIT );
856
857 DONE = 4;
858 pthread_mutex_lock( &netw -> eventbolt );
859 netw -> state = NETW_QUIT ;
860 pthread_mutex_unlock( &netw -> eventbolt );
861 }
862 else
863 {
864 Malloc( recvdmsg , N_STR , 1 );
865 if( !recvdmsg )
866 DONE = 3 ;
867 else
868 {
869 n_log( LOG_DEBUG , "%d octets to receive..." , nboctet );
870 Malloc( recvdmsg -> data , char , nboctet + 1 );
871 recvdmsg -> length = nboctet + 1 ;
872 recvdmsg -> written = nboctet ;
873
874 /* receiving the data itself */
875 if ( recv_data( netw -> link . sock , recvdmsg -> data , nboctet ) == -1 )
876 DONE = 2 ;
877
878 pthread_mutex_lock( &netw -> recvbolt );
879
880 if( !DONE && list_push( netw -> recv_buf , recvdmsg , free_nstr_ptr ) == FALSE )
881 DONE = 5 ;
882
883 pthread_mutex_unlock( &netw -> recvbolt );
884 }
885 }
886 }
887 }
888 else
889 {
890 #if !defined( LINUX ) && !defined( SOLARIS ) && !defined( AIX )
891 rest( 10 );
892 #else
893 usleep( 10000 );
894 #endif
895 }
896 }
897
898 }while( !DONE );
899
900 if( DONE == 1 )
901 n_log( LOG_ERR , "Error when receiving number of octet from socket %d" , netw -> link . sock );
902 else if( DONE == 2 )
903 n_log( LOG_ERR , "Error when receiving data from socket %d" , netw -> link . sock );
904 else if( DONE == 3 )
905 n_log( LOG_ERR , "Error allocating receved message from socket %d" , netw -> link . sock );
906 else if( DONE == 4 )
907 n_log( LOG_DEBUG , "Quit asked, closing receiving thread on socket %d" , netw -> link . sock );
908 else if( DONE == 5 )
909 n_log( LOG_ERR , "Error adding receved message from socket %d" , netw -> link . sock );
910 else
911 n_log( LOG_ERR , "Closing recv thread for socket %d\nReason unknown\nDONE = %d" , netw -> link . sock , DONE );
912
913
914 pthread_mutex_lock( &netw -> eventbolt );
915 if( DONE == 4 )
916 netw -> state = NETW_QUIT ;
917 else
918 netw -> state = NETW_ERROR ;
919 pthread_mutex_unlock( &netw -> eventbolt );
920
921 n_log( LOG_DEBUG , "Socket %d: Receive thread exited !" , netw -> link . sock );
922
923 pthread_exit( 0 );
924
925 /* suppress compiler warning */
926 #if !defined( LINUX ) && !defined( SOLARIS )
927 return NULL ;
928 #endif
929 } /* netw_recv_func(...) */
930
931
932
938 int netw_stop_thr_engine( NETWORK *netw )
939 {
940 _assert( netw , return FALSE );
941
942 n_log( LOG_DEBUG , "Network %d stop threads event" , netw -> link . sock );
943
944 pthread_mutex_lock( &netw -> eventbolt );
945 if( netw -> threaded_engine_status != NETW_THR_ENGINE_STARTED )
946 {
947 n_log( LOG_ERR , "Couldn not stop: network thread engine not started for network %p" , netw );
948 pthread_mutex_unlock( &netw -> eventbolt );
949 return FALSE ;
950 }
951
952 netw -> state = NETW_QUIT ;
953
954 pthread_mutex_unlock( &netw -> eventbolt );
955
956 n_log( LOG_DEBUG , "Network %d waits for send threads to stop" , netw -> link . sock );
957 pthread_join( netw -> send_thr , NULL );
958 n_log( LOG_DEBUG , "Network %d waits for recv threads to stop" , netw -> link . sock );
959 pthread_join( netw -> recv_thr , NULL );
960
961 pthread_mutex_lock( &netw -> eventbolt );
962 netw -> state = NETW_STOP ;
963 netw -> threaded_engine_status = NETW_THR_ENGINE_STOPPED ;
964 pthread_mutex_unlock( &netw -> eventbolt );
965
966 n_log( LOG_DEBUG , "Network %d threads Stopped !" , netw -> link . sock );
967
968 return TRUE;
969 } /* Stop_Network( ... ) */
970
971
972
980 int send_data( SOCKET s, char *buf, NSTRBYTE n )
981 {
982 int bcount = 0 ; /* counts bytes read */
983 int br = 0 ; /* bytes read this pass */
984
985 _assert( buf , return -1 );
986
987 if( n == 0 )
988 {
989 n_log( LOG_ERR , "Send of 0 is unsupported." );
990 return -1 ;
991 }
992
993 while( (NSTRBYTE)bcount < n ) /* loop until full buffer */
994 {
995 if ( ( br = send( s, buf, n - bcount, NETFLAGS ) ) > 0 )
996 {
997 bcount += br; /* increment byte counter */
998 buf += br; /* move buffer ptr for next read */
999 }
1000 else
1001 if ( br <= 0 )
1002 { /* signal an error to the caller */
1003 n_log( LOG_ERR , "Socket %d sending Error" , s );
1004 return -1 ;
1005 }
1006 }
1007
1008 return bcount;
1009 } /*send_data(...)*/
1010
1011
1012
1020 int recv_data( SOCKET s, char *buf , NSTRBYTE n )
1021 {
1022 int bcount = 0 ; /* counts bytes read */
1023 int br = 0 ; /* bytes read this pass */
1024
1025 _assert( buf , return -1 );
1026
1027 if( n == 0 )
1028 {
1029 n_log( LOG_ERR , "Recv of 0 is unsupported." );
1030 return -1 ;
1031 }
1032
1033 while( (NSTRBYTE)bcount < n )
1034 { /* loop until full buffer */
1035
1036 if ( ( br = recv( s, buf, n - bcount, NETFLAGS ) ) > 0 )
1037 {
1038 bcount += br; /* increment byte counter */
1039 buf += br; /* move buffer ptr for next read */
1040 }
1041 else
1042 if ( br <= 0 )
1043 { /* signal an error to the caller */
1044 n_log( LOG_ERR , "Socket %d receive Error" , s );
1045 return -1 ;
1046 }
1047 }
1048
1049 return bcount;
1050 } /*recv_data(...)*/