source: source/ariba/utility/transport/tcpip/protlib/tp_over_tcp.cpp@ 10640

Last change on this file since 10640 was 10432, checked in by bless@…, 13 years ago
  • added close() statement for listener socket
File size: 55.9 KB
Line 
1/// ----------------------------------------*- mode: C++; -*--
2/// @file tp_over_tcp.cpp
3/// TCP-based transport module (includes framing support)
4/// ----------------------------------------------------------
5/// $Id: tp_over_tcp.cpp 2872 2008-02-18 10:58:03Z bless $
6/// $HeadURL: https://svn.ipv6.tm.uka.de/nsis/protlib/trunk/src/tp_over_tcp.cpp $
7// ===========================================================
8//
9// Copyright (C) 2005-2007, all rights reserved by
10// - Institute of Telematics, Universitaet Karlsruhe (TH)
11//
12// More information and contact:
13// https://projekte.tm.uka.de/trac/NSIS
14//
15// This program is free software; you can redistribute it and/or modify
16// it under the terms of the GNU General Public License as published by
17// the Free Software Foundation; version 2 of the License
18//
19// This program is distributed in the hope that it will be useful,
20// but WITHOUT ANY WARRANTY; without even the implied warranty of
21// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
22// GNU General Public License for more details.
23//
24// You should have received a copy of the GNU General Public License along
25// with this program; if not, write to the Free Software Foundation, Inc.,
26// 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
27//
28// ===========================================================
29
30extern "C"
31{
32 //#define _SOCKADDR_LEN /* use BSD 4.4 sockets */
33#include <unistd.h> /* gethostname */
34#include <sys/types.h> /* network socket interface */
35#include <netinet/in.h> /* network socket interface */
36#include <netinet/tcp.h> /* for TCP Socket Option */
37#include <sys/socket.h>
38#include <arpa/inet.h> /* inet_addr */
39
40#include <fcntl.h>
41#include <sys/poll.h>
42}
43
44#include <iostream>
45#include <errno.h>
46#include <string>
47#include <sstream>
48
49#include "tp_over_tcp.h"
50#include "threadsafe_db.h"
51#include "cleanuphandler.h"
52#include "setuid.h"
53#include "queuemanager.h"
54#include "logfile.h"
55
56#include <set>
57
58#define TCP_SUCCESS 0
59#define TCP_SEND_FAILURE 1
60
61const unsigned int max_listen_queue_size= 10;
62
63#define IPV6_ADDR_INT32_SMP 0x0000ffff
64
65namespace protlib {
66
67void v6_to_v4(struct sockaddr_in *sin, struct sockaddr_in6 *sin6) {
68 bzero(sin, sizeof(*sin));
69 sin->sin_family = AF_INET;
70 sin->sin_port = sin6->sin6_port;
71 memcpy(&sin->sin_addr, &sin6->sin6_addr.s6_addr[12], sizeof(struct in_addr));
72}
73
74/* Convert sockaddr_in to sockaddr_in6 in v4 mapped addr format. */
75void v4_to_v6(struct sockaddr_in *sin, struct sockaddr_in6 *sin6) {
76 bzero(sin6, sizeof(*sin6));
77 sin6->sin6_family = AF_INET6;
78 sin6->sin6_port = sin->sin_port;
79 *(uint32_t *)&sin6->sin6_addr.s6_addr[0] = 0;
80 *(uint32_t *)&sin6->sin6_addr.s6_addr[4] = 0;
81 *(uint32_t *)&sin6->sin6_addr.s6_addr[8] = IPV6_ADDR_INT32_SMP;
82 *(uint32_t *)&sin6->sin6_addr.s6_addr[12] = sin->sin_addr.s_addr;
83}
84
85using namespace log;
86
87/** @defgroup protlib
88 * @ingroup protlib
89 * @{
90 */
91
92char in6_addrstr[INET6_ADDRSTRLEN+1];
93
94
95/******* class TPoverTCP *******/
96
97
98/** get_connection_to() checks for already existing connections.
99 * If a connection exists, it returns "AssocData"
100 * and saves it in "connmap" for further use
101 * If no connection exists, a new connection to "addr"
102 * is created.
103 */
104AssocData*
105TPoverTCP::get_connection_to(const appladdress& addr)
106{
107 // get timeout
108 struct timespec ts;
109 get_time_of_day(ts);
110 ts.tv_nsec+= tpparam.sleep_time * 1000000L;
111 if (ts.tv_nsec>=1000000000L)
112 {
113 ts.tv_sec += ts.tv_nsec / 1000000000L;
114 ts.tv_nsec= ts.tv_nsec % 1000000000L;
115 }
116
117 // create a new AssocData pointer, initialize it to NULL
118 AssocData* assoc= NULL;
119 int new_socket;
120 // loop until timeout is exceeded: TODO: check applicability of loop
121 do
122 {
123 // check for existing connections to addr
124 // start critical section
125 lock(); // install_cleanup_thread_lock(TPoverTCP, this);
126 assoc= connmap.lookup(addr);
127 // end critical section
128 unlock(); // uninstall_cleanup(1);
129 if (assoc)
130 {
131 // If not shut down then use it, else abort, wait and check again.
132 if (!assoc->shutdown)
133 {
134 return assoc;
135 }
136 else
137 {
138 // TODO: connection is already in shutdown mode. What now?
139 ERRCLog(tpparam.name,"socket exists, but is already in mode shutdown");
140
141 return 0;
142 }
143 } //end __if (assoc)__
144 else
145 {
146 Log(DEBUG_LOG,LOG_UNIMP,tpparam.name,"No existing connection to "
147 << addr.get_ip_str() << " port #" << addr.get_port() << " found, creating a new one.");
148 }
149
150 // no connection found, create a new one
151 new_socket = socket( v4_mode ? AF_INET : AF_INET6, SOCK_STREAM, IPPROTO_TCP);
152 if (new_socket == -1)
153 {
154 ERRCLog(tpparam.name,"Couldn't create a new socket: " << strerror(errno));
155
156 return 0;
157 }
158
159 // Disable Nagle Algorithm, set (TCP_NODELAY)
160 int nodelayflag= 1;
161 int status= setsockopt(new_socket,
162 IPPROTO_TCP,
163 TCP_NODELAY,
164 &nodelayflag,
165 sizeof(nodelayflag));
166 if (status)
167 {
168 ERRLog(tpparam.name, "Could not set socket option TCP_NODELAY:" << strerror(errno));
169 }
170
171 // Reuse ports, even if they are busy
172 int socketreuseflag= 1;
173 status= setsockopt(new_socket,
174 SOL_SOCKET,
175 SO_REUSEADDR,
176 (const char *) &socketreuseflag,
177 sizeof(socketreuseflag));
178 if (status)
179 {
180 ERRLog(tpparam.name, "Could not set socket option SO_REUSEADDR:" << strerror(errno));
181 }
182
183 struct sockaddr_in6 dest_address;
184 dest_address.sin6_flowinfo= 0;
185 dest_address.sin6_scope_id= 0;
186 addr.get_sockaddr(dest_address);
187
188 // connect the socket to the destination address
189 int connect_status = 0;
190 if (v4_mode) {
191 struct sockaddr_in dest_address_v4;
192 v6_to_v4( &dest_address_v4, &dest_address );
193 connect_status = connect(new_socket,
194 reinterpret_cast<const struct sockaddr*>(&dest_address_v4),
195 sizeof(dest_address));
196 } else {
197 connect_status = connect(new_socket,
198 reinterpret_cast<const struct sockaddr*>(&dest_address),
199 sizeof(dest_address));
200 }
201
202 // connects to the listening_port of the peer's masterthread
203 if (connect_status != 0)
204 {
205 ERRLog(tpparam.name,"Connect to " << addr.get_ip_str() << " port #" << addr.get_port()
206 << " failed: [" << color[red] << strerror(errno) << color[off] << "]");
207
208 close(new_socket);
209 return 0; // error: couldn't connect
210 }
211
212
213 struct sockaddr_in6 own_address;
214 if (v4_mode) {
215 struct sockaddr_in own_address_v4;
216 socklen_t own_address_len_v4 = sizeof(own_address_v4);
217 getsockname(new_socket, reinterpret_cast<struct sockaddr*>(&own_address_v4), &own_address_len_v4);
218 v4_to_v6(&own_address_v4, &own_address);
219 } else {
220 socklen_t own_address_len= sizeof(own_address);
221 getsockname(new_socket, reinterpret_cast<struct sockaddr*>(&own_address), &own_address_len);
222 }
223
224 Log(DEBUG_LOG,LOG_UNIMP, tpparam.name,">>--Connect-->> to " << addr.get_ip_str() << " port #" << addr.get_port()
225 << " from " << inet_ntop(AF_INET6,&own_address.sin6_addr,in6_addrstr,INET6_ADDRSTRLEN)
226 << " port #" << ntohs(own_address.sin6_port));
227
228
229 // create new AssocData record (will copy addr)
230 assoc = new(nothrow) AssocData(new_socket, addr, appladdress(own_address,IPPROTO_TCP));
231
232 // if assoc could be successfully created, insert it into ConnectionMap
233 if (assoc)
234 {
235 bool insert_success= false;
236 // start critical section
237 lock(); // install_cleanup_thread_lock(TPoverTCP, this);
238 // insert AssocData into connection map
239 insert_success= connmap.insert(assoc);
240 // end critical section
241 unlock(); // uninstall_cleanup(1);
242
243 if (insert_success == true)
244 {
245#ifdef _DEBUG
246 Log(DEBUG_LOG,LOG_UNIMP, tpparam.name, "Connected to " << addr.get_ip_str() << ", port #" << addr.get_port()
247 << " via socket " << new_socket);
248
249
250#endif
251
252 // notify master thread to start a receiver thread (i.e. send selfmessage)
253 TPoverTCPMsg* newmsg= new(nothrow)TPoverTCPMsg(assoc, tpparam.source, TPoverTCPMsg::start);
254 if (newmsg)
255 {
256 bool ret = newmsg->send_to(tpparam.source);
257 if(!ret) delete newmsg;
258 return assoc;
259 }
260 else
261 ERRCLog(tpparam.name,"get_connection_to: could not get memory for internal msg");
262 }
263 else
264 {
265 // delete data and abort
266 ERRCLog(tpparam.name, "Cannot insert AssocData for socket " << new_socket << ", "<< addr.get_ip_str()
267 <<", port #" << addr.get_port() << " into connection map, aborting connection");
268
269 // abort connection, delete its AssocData
270 close (new_socket);
271 if (assoc)
272 {
273 delete assoc;
274 assoc= 0;
275 }
276 return assoc;
277 } // end else connmap.insert
278
279 } // end "if (assoc)"
280 }
281 while (wait_cond(ts)!=ETIMEDOUT);
282
283 return assoc;
284} //end get_connection_to
285
286
287/** terminates a signaling association/connection
288 * - if no connection exists, generate a warning
289 * - otherwise: generate internal msg to related receiver thread
290 */
291void
292TPoverTCP::terminate(const address& in_addr)
293{
294#ifndef _NO_LOGGING
295 const char *const thisproc="terminate() - ";
296#endif
297
298 appladdress* addr = NULL;
299 addr = dynamic_cast<appladdress*>(in_addr.copy());
300
301 if (!addr) return;
302
303 // Create a new AssocData-pointer
304 AssocData* assoc = NULL;
305
306 // check for existing connections to addr
307
308 // start critical section:
309 // please note if receiver_thread terminated already, the assoc data is not
310 // available anymore therefore we need a lock around cleanup_receiver_thread()
311
312 lock(); // install_cleanup_thread_lock(TPoverTCP, this);
313 assoc= connmap.lookup(*addr);
314 if (assoc)
315 {
316 EVLog(tpparam.name,thisproc<<"got request to shutdown connection for peer " << addr);
317 // If not shut down then use it, else abort, wait and check again.
318 if (!assoc->shutdown)
319 {
320 if (assoc->socketfd)
321 {
322 // disallow sending
323 if (shutdown(assoc->socketfd,SHUT_WR))
324 {
325 ERRLog(tpparam.name,thisproc<<"shutdown (write) on socket for peer " << addr << " returned error:" << strerror(errno));
326 }
327 else
328 {
329 EVLog(tpparam.name,thisproc<<"initiated closing of connection for peer " << addr << ". Shutdown (write) on socket "<< assoc->socketfd );
330 }
331 }
332 assoc->shutdown= true;
333 }
334 else
335 EVLog(tpparam.name,thisproc<<"connection for peer " << addr << " is already in mode shutdown");
336
337 }
338 else
339 WLog(tpparam.name,thisproc<<"could not find a connection for peer " << *addr);
340
341 stop_receiver_thread(assoc);
342
343 // end critical section
344 unlock(); // uninstall_cleanup(1);
345
346 if (addr) delete addr;
347}
348
349
350/** generates and internal TPoverTCP message to send a NetMsg to the network
351 * - it is necessary to let a thread do this, because the caller
352 * may get blocked if the connect() or send() call hangs for a while
353 * - the sending thread will call TPoverTCP::tcpsend()
354 * - if no connection exists, creates a new one (unless use_existing_connection is true)
355 * @note the netmsg is deleted by the send() method when it is not used anymore
356 */
357void
358TPoverTCP::send(NetMsg* netmsg, const address& in_addr, bool use_existing_connection)
359{
360 if (netmsg == NULL) {
361 ERRCLog(tpparam.name,"send() - called without valid NetMsg buffer (NULL)");
362 return;
363 }
364
365 appladdress* addr = NULL;
366 addr= dynamic_cast<appladdress*>(in_addr.copy());
367
368 if (!addr)
369 {
370 ERRCLog(tpparam.name,"send() - given destination address is not of expected type (appladdress), has type " << (int) in_addr.get_type());
371 return;
372 }
373
374 // lock due to sendermap access
375 lock();
376
377 // check for existing sender thread
378 sender_thread_queuemap_t::const_iterator it= senderthread_queuemap.find(*addr);
379
380 FastQueue* destqueue= 0;
381
382 if (it == senderthread_queuemap.end())
383 { // no sender thread found so far
384
385 // if we already have an existing connection it is save to create a sender thread
386 // since get_connection_to() will not be invoked, so an existing connection will
387 // be used
388 const AssocData* assoc = connmap.lookup(*addr);
389
390 //DLog(tpparam.name,"send() - use_existing_connection:" << (use_existing_connection ? "true" : "false") << " assoc:" << assoc);
391
392 if (use_existing_connection==false || (assoc && assoc->shutdown==false && assoc->socketfd>0))
393 { // it is allowed to create a new connection for this thread
394 // create a new queue for sender thread
395 FastQueue* sender_thread_queue= new FastQueue;
396 create_new_sender_thread(sender_thread_queue);
397 // remember queue for later use
398
399 //pair<sender_thread_queuemap_t::iterator, bool> tmpinsiterator=
400 senderthread_queuemap.insert( pair<appladdress,FastQueue*> (*addr,sender_thread_queue) );
401
402 destqueue= sender_thread_queue;
403 }
404 }
405 else
406 { // we have a sender thread, use stored queue from map
407 destqueue= it->second;
408 }
409
410 unlock();
411
412 // send a send_data message to it (if we have a destination queue)
413 if (destqueue)
414 {
415 // both parameters will be freed after message was sent!
416
417 appladdress* apl=new appladdress(*addr);
418 TPoverTCPMsg* internalmsg= new TPoverTCPMsg(netmsg,apl);
419 if (internalmsg)
420 {
421 // send the internal message to the sender thread queue
422 bool sent = internalmsg->send(tpparam.source,destqueue);
423 if (!sent) {
424 delete internalmsg->get_appladdr();
425 delete internalmsg;
426 internalmsg = NULL;
427 }
428 }
429 }
430 else
431 {
432 if (!use_existing_connection)
433 WLog(tpparam.name,"send() - found entry for address, but no active sender thread available for peer addr:" << *addr << " - dropping data");
434 else
435 {
436 DLog(tpparam.name,"no active sender thread found for peer " << *addr << " - but policy forbids to set up a new connection, will drop data");
437
438 }
439 // cannot send data, so we must drop it
440 delete netmsg;
441 }
442
443 if (addr) delete addr;
444}
445
446/** sends a NetMsg to the network.
447 *
448 * @param netmsg message to send
449 * @param addr transport endpoint address
450 *
451 * @note if no connection exists, creates a new one
452 * @note both parameters are deleted after the message was sent
453 */
454void
455TPoverTCP::tcpsend(NetMsg* netmsg, appladdress* addr)
456{
457#ifndef _NO_LOGGING
458 const char *const thisproc="sender - ";
459#endif
460
461 // set result initially to success, set it to failure
462 // in places where these failures occur
463 int result = TCP_SUCCESS;
464 int saved_errno= 0;
465 int ret= 0;
466
467 // Create a new AssocData-pointer
468 AssocData* assoc = NULL;
469
470 // tp.cpp checks for initialisation of tp and correctness of
471 // msgsize, protocol and ip,
472 // throws an error if something is not right
473 if (addr) {
474 addr->convert_to_ipv6();
475 check_send_args(*netmsg,*addr);
476 }
477 else
478 {
479 ERRCLog(tpparam.name, thisproc << "address pointer is NULL");
480 result= TCP_SEND_FAILURE;
481
482 delete netmsg;
483 delete addr;
484
485 throw TPErrorInternal();
486 }
487
488
489 // check for existing connections,
490 // if a connection exists, return its AssocData
491 // and save it in assoc for further use
492 // if no connection exists, create a new one (in get_connection_to()).
493 // Connection is inserted into connection map that must be done
494 // with exclusive access
495 assoc= get_connection_to(*addr);
496
497 if (assoc==NULL || assoc->socketfd<=0)
498 {
499 ERRCLog(tpparam.name, color[red] << thisproc << "no valid assoc/socket data - dropping packet");
500
501 delete netmsg;
502 delete addr;
503 return;
504 }
505
506 if (assoc->shutdown)
507 {
508 Log(WARNING_LOG, LOG_ALERT, tpparam.name, thisproc << "should send message although connection already half closed");
509 delete netmsg;
510 delete addr;
511
512 throw TPErrorSendFailed();
513 }
514
515 uint32 msgsize= netmsg->get_size();
516#ifdef DEBUG_HARD
517 cerr << thisproc << "message size=" << netmsg->get_size() << endl;
518#endif
519
520 const uint32 retry_send_max = 3;
521 uint32 retry_count = 0;
522 // send all the data contained in netmsg to the socket
523 // which belongs to the address "addr"
524 for (uint32 bytes_sent= 0;
525 bytes_sent < msgsize;
526 bytes_sent+= ret)
527 {
528
529#ifdef _DEBUG_HARD
530 for (uint32 i=0;i<msgsize;i++)
531 {
532 cout << "send_buf: " << i << " : ";
533 if ( isalnum(*(netmsg->get_buffer()+i)) )
534 cout << "'" << *(netmsg->get_buffer()+i) << "' (0x" << hex << (unsigned short) *(netmsg->get_buffer()+i) << dec << ")" ;
535 else
536 cout << "0x" << hex << (unsigned short) *(netmsg->get_buffer()+i) << dec;
537 cout << endl;
538 }
539
540 cout << endl;
541 cout << "bytes_sent: " << bytes_sent << endl;
542 cout << "Message size: " << msgsize << endl;
543 cout << "Send-Socket: " << assoc->socketfd << endl;
544 cout << "pointer-Offset. " << netmsg->get_pos() << endl;
545 cout << "vor send " << endl;
546#endif
547
548 retry_count= 0;
549 do
550 {
551 // socket send
552 ret= ::send(assoc->socketfd,
553 netmsg->get_buffer() + bytes_sent,
554 msgsize - bytes_sent,
555 MSG_NOSIGNAL);
556
557 // send_buf + bytes_sent
558
559 // Deal with temporary internal errors like EAGAIN, resource temporary unavailable etc.
560 // retry sending
561 if (ret < 0)
562 { // internal error during send occured
563 saved_errno= errno;
564 switch(saved_errno)
565 {
566 case EAGAIN: // same as EWOULDBLOCK
567 case EINTR: // man page says: A signal occurred before any data was transmitted.
568 case ENOBUFS: // The output queue for a network interface was full.
569 retry_count++;
570 ERRLog(tpparam.name,"Temporary failure while calling send(): " << strerror(saved_errno) << ", errno: " << saved_errno
571 << " - retry sending, retry #" << retry_count);
572 // pace down a little bit, sleep for a while
573 sleep(1);
574 break;
575
576 // everything else should not lead to repetition
577 default:
578 retry_count= retry_send_max;
579 break;
580 } // end switch
581 } // endif
582 else // leave while
583 break;
584 }
585 while(retry_count < retry_send_max);
586
587 if (ret < 0)
588 { // unrecoverable error occured
589
590 result= TCP_SEND_FAILURE;
591 break;
592 } // end if (ret < 0)
593 else
594 {
595 if (debug_pdu)
596 {
597 ostringstream hexdump;
598 netmsg->hexdump(hexdump,netmsg->get_buffer(),bytes_sent);
599 DLog(tpparam.name,"PDU debugging enabled - Sent:" << hexdump.str());
600 }
601 }
602
603 // continue with send next data block in next for() iteration
604 } // end for
605
606 // *** note: netmsg is deleted here ***
607 delete netmsg;
608
609 // Throwing an exception within a critical section does not
610 // unlock the mutex.
611
612 if (result != TCP_SUCCESS)
613 {
614 ERRLog(tpparam.name, thisproc << "TCP error, returns " << ret << ", error : " << strerror(errno));
615 delete addr;
616
617 throw TPErrorSendFailed(saved_errno);
618
619 }
620 else
621 EVLog(tpparam.name, thisproc << ">>----Sent---->> message (" << msgsize << " bytes) using socket " << assoc->socketfd << " to " << *addr);
622
623 if (!assoc) {
624 // no connection
625
626 ERRLog(tpparam.name, thisproc << "cannot get connection to " << addr->get_ip_str()
627 << ", port #" << addr->get_port());
628
629 delete addr;
630
631 throw TPErrorUnreachable(); // should be no assoc found
632 } // end "if (!assoc)"
633
634 // *** delete address ***
635 delete addr;
636}
637
638/* this thread waits for an internal message that either:
639 * - requests transmission of a packet
640 * - requests to stop this thread
641 * @param argp points to the thread queue for internal messages
642 */
643void
644TPoverTCP::sender_thread(void *argp)
645{
646#ifndef _NO_LOGGING
647 const char *const methodname="senderthread - ";
648#endif
649
650 message* internal_thread_msg = NULL;
651
652 EVLog(tpparam.name, methodname << "starting as thread <" << pthread_self() << ">");
653
654 FastQueue* fq= reinterpret_cast<FastQueue*>(argp);
655 if (!fq)
656 {
657 ERRLog(tpparam.name, methodname << "thread <" << pthread_self() << "> no valid pointer to msg queue. Stop.");
658 return;
659 }
660
661 bool terminate= false;
662 TPoverTCPMsg* internalmsg= 0;
663 while (terminate==false && (internal_thread_msg= fq->dequeue()) != 0 )
664 {
665 internalmsg= dynamic_cast<TPoverTCPMsg*>(internal_thread_msg);
666
667 if (internalmsg == 0)
668 {
669 ERRLog(tpparam.name, methodname << "received not an TPoverTCPMsg but a" << internal_thread_msg->get_type_name());
670 }
671 else
672 if (internalmsg->get_msgtype() == TPoverTCPMsg::send_data)
673 {
674 // create a connection if none exists and send the netmsg
675 if (internalmsg->get_netmsg() && internalmsg->get_appladdr())
676 {
677 try
678 {
679 tcpsend(internalmsg->get_netmsg(),internalmsg->get_appladdr());
680 } // end try
681 catch(TPErrorSendFailed& err)
682 {
683 ERRLog(tpparam.name, methodname << "TCP send call failed - " << err.what()
684 << " cause: (" << err.get_reason() << ") " << strerror(err.get_reason()) );
685 } // end catch
686 catch(TPError& err)
687 {
688 ERRLog(tpparam.name, methodname << "TCP send call failed - reason: " << err.what());
689 } // end catch
690 catch(...)
691 {
692 ERRLog(tpparam.name, methodname << "TCP send call failed - unknown exception");
693 }
694 }
695 else
696 {
697 ERRLog(tpparam.name, methodname << "problem with passed arguments references, they point to 0");
698 }
699 }
700 else
701 if (internalmsg->get_msgtype() == TPoverTCPMsg::stop)
702 {
703 terminate= true;
704 }
705
706 delete internalmsg;
707 } // end while
708
709 EVLog(tpparam.name, methodname << "<" << pthread_self() << "> terminated connection.");
710}
711
712
713/** receiver thread listens at a TCP socket for incoming PDUs
714 * and passes complete PDUs to the coordinator. Incomplete
715 * PDUs due to aborted connections or buffer overflows are discarded.
716 * @param argp - assoc data and flags sig_terminate and terminated
717 *
718 * @note this is a static member function, so you cannot use class variables
719 */
720void
721TPoverTCP::receiver_thread(void *argp)
722{
723#ifndef _NO_LOGGING
724 const char *const methodname="receiver - ";
725#endif
726
727 receiver_thread_arg_t *receiver_thread_argp= static_cast<receiver_thread_arg_t *>(argp);
728 const appladdress* peer_addr = NULL;
729 const appladdress* own_addr = NULL;
730 uint32 bytes_received = 0;
731 TPMsg* tpmsg= NULL;
732
733 // argument parsing - start
734 if (receiver_thread_argp == 0)
735 {
736 ERRCLog(tpparam.name, methodname << "No arguments given at start of receiver thread <" << pthread_self() << ">, exiting.");
737
738 return;
739 }
740 else
741 {
742 // change status to running, i.e., not terminated
743 receiver_thread_argp->terminated= false;
744
745#ifdef _DEBUG
746 DLog(tpparam.name, methodname << "New receiver thread <" << pthread_self() << "> started. ");
747#endif
748 }
749
750 int conn_socket= 0;
751 if (receiver_thread_argp->peer_assoc)
752 {
753 // get socket descriptor from arg
754 conn_socket = receiver_thread_argp->peer_assoc->socketfd;
755 // get pointer to peer address of socket (source/sender address of peer) from arg
756 peer_addr= &receiver_thread_argp->peer_assoc->peer;
757 own_addr= &receiver_thread_argp->peer_assoc->ownaddr;
758 }
759 else
760 {
761 ERRCLog(tpparam.name, methodname << "No peer assoc available - pointer is NULL");
762
763 return;
764 }
765
766 if (peer_addr == 0)
767 {
768 ERRCLog(tpparam.name, methodname << "No peer address available for socket " << conn_socket << ", exiting.");
769
770 return;
771 }
772 // argument parsing - end
773#ifdef _DEBUG
774 Log(DEBUG_LOG,LOG_UNIMP, tpparam.name, methodname <<
775 "Preparing to wait for data at socket "
776 << conn_socket << " from " << receiver_thread_argp->peer_assoc->peer);
777#endif
778
779 int ret= 0;
780 uint32 msgcontentlength= 0;
781 bool msgcontentlength_known= false;
782 bool pdu_complete= false; // when to terminate inner loop
783
784 /* maybe use this to create a new pdu,
785 /// constructor
786 contextlistpdu(type_t t, subtype_t st, uint32 fc, uint32 numobj = 0);
787 */
788
789 // activate O_NON_BLOCK for recv on socket conn_socket
790 // CAVEAT: this also implies non-blocking send()!
791 //fcntl(conn_socket,F_SETFL, O_NONBLOCK);
792
793 // set options for polling
794 const unsigned int number_poll_sockets= 1;
795 struct pollfd poll_fd;
796 // have to set structure before poll call
797 poll_fd.fd = conn_socket;
798 poll_fd.events = POLLIN | POLLPRI;
799 poll_fd.revents = 0;
800
801 int poll_status;
802 bool recv_error= false;
803
804 NetMsg* netmsg= 0;
805 NetMsg* remainbuf= 0;
806 size_t buffer_bytes_left= 0;
807 size_t trailingbytes= 0;
808 bool skiprecv= false;
809 // loop until we receive a terminate signal (read-only var for this thread)
810 // or get an error from socket read
811 while( receiver_thread_argp->sig_terminate == false )
812 {
813 // Read next PDU from socket or process trailing bytes in remainbuf
814 ret= 0;
815 msgcontentlength= 0;
816 msgcontentlength_known= false;
817 pdu_complete= false;
818 netmsg= 0;
819
820 // there are trailing bytes left from the last receive call
821 if (remainbuf != 0)
822 {
823 netmsg= remainbuf;
824 remainbuf= 0;
825 buffer_bytes_left= netmsg->get_size()-trailingbytes;
826 bytes_received= trailingbytes;
827 trailingbytes= 0;
828 skiprecv= true;
829 }
830 else // no trailing bytes, create a new buffer
831 if ( (netmsg= new NetMsg(NetMsg::max_size)) != 0 )
832 {
833 buffer_bytes_left= netmsg->get_size();
834 bytes_received= 0;
835 skiprecv= false;
836 }
837 else
838 { // buffer allocation failed
839 bytes_received= 0;
840 buffer_bytes_left= 0;
841 recv_error= true;
842 }
843
844 // loops until PDU is complete
845 // >>>>>>>>>>>>>>>>>>>>>>>>>>> while >>>>>>>>>>>>>>>>>>>>>>>>
846 while (!pdu_complete &&
847 !recv_error &&
848 !receiver_thread_argp->sig_terminate)
849 {
850 if (!skiprecv)
851 {
852 // read from TCP socket or return after sleep_time
853 poll_status= poll(&poll_fd, number_poll_sockets, tpparam.sleep_time);
854
855 if (receiver_thread_argp->sig_terminate)
856 {
857 Log(EVENT_LOG,LOG_UNIMP,tpparam.name,methodname << "Thread <" << pthread_self() << "> found terminate signal after poll");
858 // disallow sending
859 AssocData* myassoc=const_cast<AssocData *>(receiver_thread_argp->peer_assoc);
860 if (myassoc->shutdown == false)
861 {
862 myassoc->shutdown= true;
863 if (shutdown(myassoc->socketfd,SHUT_WR))
864 {
865 if ( errno != ENOTCONN )
866 Log(ERROR_LOG,LOG_UNIMP,tpparam.name,methodname <<"shutdown (write) on socket " << conn_socket << " returned error:" << strerror(errno));
867 }
868 }
869 // try to read do a last read from the TCP socket or return after sleep_time
870 if (poll_status == 0)
871 {
872 poll_status= poll(&poll_fd, number_poll_sockets, tpparam.sleep_time);
873 }
874 }
875
876 if (poll_fd.revents & POLLERR) // Error condition
877 {
878 if (errno == 0 || errno == EINTR)
879 {
880 EVLog(tpparam.name, methodname << "poll(): " << strerror(errno));
881 }
882 else
883 {
884 ERRCLog(tpparam.name, methodname << "Poll indicates error: " << strerror(errno));
885 recv_error= true;
886 }
887 }
888
889 if (poll_fd.revents & POLLHUP) // Hung up
890 {
891 Log(EVENT_LOG,LOG_CRIT, tpparam.name, methodname << "Poll hung up");
892 recv_error= true;
893 }
894
895 if (poll_fd.revents & POLLNVAL) // Invalid request: fd not open
896 {
897 EVLog(tpparam.name, methodname << "Poll Invalid request: fd not open");
898 recv_error= true;
899 }
900
901 // check status (return value) of poll call
902 switch (poll_status)
903 {
904 case -1:
905 if (errno == 0 || errno == EINTR)
906 {
907 EVLog(tpparam.name, methodname << "Poll status: " << strerror(errno));
908 }
909 else
910 {
911 ERRCLog(tpparam.name, methodname << "Poll status indicates error: " << strerror(errno) << "- aborting");
912 recv_error= true;
913 }
914
915 continue; // next while iteration
916 break;
917
918 case 0:
919#ifdef DEBUG_HARD
920 Log(DEBUG_LOG,LOG_UNIMP, tpparam.name, methodname << "Poll timed out after " << tpparam.sleep_time << " ms.");
921#endif
922 continue; // next while iteration
923 break;
924
925 default:
926#ifdef DEBUG_HARD
927 Log(DEBUG_LOG,LOG_UNIMP, tpparam.name, methodname << "Poll: " << poll_status << " event(s) ocurred, of type " << poll_fd.revents);
928#endif
929 break;
930 } // end switch
931
932
933 /// receive data from socket buffer (recv will not block)
934 ret = recv(conn_socket,
935 netmsg->get_buffer() + bytes_received,
936 buffer_bytes_left,
937 MSG_DONTWAIT);
938
939 if ( ret < 0 )
940 {
941 delete netmsg;
942 if (errno!=EAGAIN && errno!=EWOULDBLOCK)
943 {
944 ERRCLog(tpparam.name, methodname << "Receive at socket " << conn_socket << " failed, error: " << strerror(errno));
945 recv_error= true;
946 continue;
947 }
948 else
949 { // errno==EAGAIN || errno==EWOULDBLOCK
950 // just nothing to read from socket, continue w/ next poll
951 continue;
952 }
953 }
954 else
955 {
956 if (ret == 0)
957 {
958 // this means that EOF is reached,
959 // other side has closed connection
960 Log(DEBUG_LOG,LOG_UNIMP, tpparam.name, methodname << "Other side (" << *peer_addr << ") closed connection for socket " << conn_socket);
961 // disallow sending
962 AssocData* myassoc=const_cast<AssocData *>(receiver_thread_argp->peer_assoc);
963 if (myassoc->shutdown == false)
964 {
965 myassoc->shutdown= true;
966 if (shutdown(myassoc->socketfd,SHUT_WR))
967 {
968 if ( errno != ENOTCONN )
969 Log(ERROR_LOG,LOG_UNIMP,tpparam.name, methodname << "shutdown (write) on socket " << conn_socket << " returned error:" << strerror(errno));
970 }
971 }
972 // not a real error, but we must quit the receive loop
973 recv_error= true;
974 }
975 else
976 {
977
978 Log(EVENT_LOG,LOG_UNIMP, tpparam.name, methodname << "<<--Received--<< packet (" << ret << " bytes) at socket " << conn_socket << " from " << *peer_addr);
979 // track number of received bytes
980 bytes_received+= ret;
981 buffer_bytes_left-= ret;
982 }
983 }
984 } // end if do not skip recv() statement
985
986 if (buffer_bytes_left < 0) ///< buffer space exhausted now
987 {
988 recv_error= true;
989 Log(ERROR_LOG,LOG_CRIT, tpparam.name, methodname << "during receive buffer space exhausted");
990 }
991
992 if (!msgcontentlength_known) ///< common header not parsed
993 {
994 // enough bytes read to parse common header?
995 if (bytes_received >= common_header_length)
996 {
997 // get message content length in number of bytes
998 if (getmsglength(*netmsg, msgcontentlength))
999 msgcontentlength_known= true;
1000 else
1001 {
1002 ERRCLog(tpparam.name, methodname << "Not a valid protocol header - discarding received packet. received size " << msgcontentlength);
1003
1004 ostringstream hexdumpstr;
1005 netmsg->hexdump(hexdumpstr,netmsg->get_buffer(),bytes_received);
1006 DLog(tpparam.name,"dumping received bytes:" << hexdumpstr.str());
1007
1008 // reset all counters
1009 msgcontentlength= 0;
1010 msgcontentlength_known= false;
1011 bytes_received= 0;
1012 pdu_complete= false;
1013 continue;
1014 }
1015 }
1016 } // endif common header not parsed
1017
1018 // check whether we have read the whole Protocol PDU
1019 DLog(tpparam.name, "bytes_received-common_header_length=" << bytes_received-common_header_length << " msgcontentlength: " << msgcontentlength);
1020 if (msgcontentlength_known)
1021 {
1022 if (bytes_received-common_header_length >= msgcontentlength )
1023 {
1024 pdu_complete= true;
1025 // truncate buffer exactly at common_header_length+msgcontentlength==PDU size, trailing stuff is handled separately
1026 netmsg->truncate(common_header_length+msgcontentlength);
1027
1028 // trailing bytes are copied into new buffer
1029 if (bytes_received-common_header_length > msgcontentlength)
1030 {
1031 WLog(tpparam.name,"trailing bytes - received more bytes ("<<bytes_received<<") than expected for PDU (" << common_header_length+msgcontentlength << ")");
1032 remainbuf= new NetMsg(NetMsg::max_size);
1033 trailingbytes= (bytes_received-common_header_length) - msgcontentlength;
1034 bytes_received= common_header_length+msgcontentlength;
1035 memcpy(remainbuf->get_buffer(),netmsg->get_buffer()+common_header_length+msgcontentlength, trailingbytes);
1036 }
1037 }
1038 else
1039 { // not enough bytes in the buffer must continue to read from network
1040 skiprecv= false;
1041 }
1042 } // endif msgcontentlength_known
1043 } // end while (!pdu_complete && !recv_error && !signalled for termination)
1044 // >>>>>>>>>>>>>>>>>>>>>>>>>>> while >>>>>>>>>>>>>>>>>>>>>>>>
1045
1046 // if other side closed the connection, we should still be able to deliver the remaining data
1047 if (ret == 0)
1048 {
1049 recv_error= false;
1050 }
1051
1052 // deliver only complete PDUs to signaling module
1053 if (!recv_error && pdu_complete)
1054 {
1055 // create TPMsg and send it to the signaling thread
1056 tpmsg = new(nothrow) TPMsg(netmsg, peer_addr->copy(), own_addr->copy());
1057 if (tpmsg)
1058 {
1059 DLog(tpparam.name, methodname << "receipt of PDU now complete, sending msg#" << tpmsg->get_id()
1060 << " to module " << message::get_qaddr_name(tpparam.dest));
1061 }
1062
1063 debug_pdu=false;
1064
1065 if (debug_pdu)
1066 {
1067 ostringstream hexdump;
1068 netmsg->hexdump(hexdump,netmsg->get_buffer(),bytes_received);
1069 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name,"PDU debugging enabled - Received:" << hexdump.str());
1070 }
1071
1072 // send the message if it was successfully created
1073 // bool message::send_to(qaddr_t dest, bool exp = false);
1074 if (!tpmsg
1075 || (!tpmsg->get_peeraddress())
1076 || (!tpmsg->send(message::qaddr_tp_over_tcp, tpparam.dest)))
1077 {
1078 Log(ERROR_LOG,LOG_NORMAL, tpparam.name, methodname << "Cannot allocate/send TPMsg");
1079 if (tpmsg) delete tpmsg;
1080 } // end if tpmsg not allocated or not addr or not sent
1081
1082
1083 } // end if !recv_error
1084 else
1085 { // error during receive or PDU incomplete
1086 if (bytes_received>0)
1087 {
1088 Log(WARNING_LOG,LOG_NORMAL, tpparam.name, methodname << "Attention! " << (recv_error? "Receive error, " : "") << (pdu_complete ? "PDU complete" : "PDU incomplete") << "received bytes: " << bytes_received);
1089 }
1090
1091 if (!pdu_complete && bytes_received>0 && bytes_received<common_header_length)
1092 {
1093 ostringstream hexdumpstr;
1094 netmsg->hexdump(hexdumpstr,netmsg->get_buffer(),bytes_received);
1095 Log(DEBUG_LOG,LOG_NORMAL,tpparam.name,"Message too short to be a valid protocol header - dumping received bytes:" << hexdumpstr.str());
1096 }
1097 // leave the outer loop
1098 /**********************/
1099 break;
1100 /**********************/
1101 } // end else
1102
1103 } // end while (thread not signalled for termination)
1104
1105 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name, methodname << "Thread <" << pthread_self()
1106 << "> shutting down and closing socket " << receiver_thread_argp->peer_assoc->peer);
1107
1108 // shutdown socket
1109 if (shutdown(conn_socket, SHUT_RD))
1110 {
1111 if ( errno != ENOTCONN )
1112 Log(ERROR_LOG,LOG_NORMAL, tpparam.name, methodname << "Thread <" << pthread_self() << "> shutdown (read) on socket failed, reason: " << strerror(errno));
1113 }
1114
1115 // close socket
1116 close(conn_socket);
1117
1118 receiver_thread_argp->terminated= true;
1119 //delete netmsg;
1120
1121 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name, methodname << "Thread <" << pthread_self() << "> terminated");
1122
1123#ifdef _DEBUG
1124 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name, methodname << "Signaling main loop for cleanup");
1125#endif
1126 // notify master thread for invocation of cleanup procedure
1127 TPoverTCPMsg* newmsg= new(nothrow)TPoverTCPMsg(receiver_thread_argp->peer_assoc);
1128 // send message to main loop thread
1129 newmsg->send_to(tpparam.source);
1130
1131}
1132
1133
1134/** this signals a terminate to a thread and wait for the thread to stop
1135 * @note it is not safe to access any thread related data after this method returned,
1136 * because the receiver thread will initiate a cleanup_receiver_thread() method
1137 * which may erase all relevant thread data.
1138 */
1139void
1140TPoverTCP::stop_receiver_thread(AssocData* peer_assoc)
1141{
1142 // All operations on recv_thread_argmap and connmap require an already acquired lock
1143 // after this procedure peer_assoc may be invalid because it was erased
1144
1145 // start critical section
1146
1147 if (peer_assoc == 0)
1148 return;
1149
1150 pthread_t thread_id= peer_assoc->thread_ID;
1151
1152 // try to clean up receiver_thread_arg
1153 recv_thread_argmap_t::iterator recv_thread_arg_iter= recv_thread_argmap.find(thread_id);
1154 receiver_thread_arg_t* recv_thread_argp=
1155 (recv_thread_arg_iter != recv_thread_argmap.end()) ? recv_thread_arg_iter->second : 0;
1156 if (recv_thread_argp)
1157 {
1158 if (!recv_thread_argp->terminated)
1159 {
1160 // thread signaled termination, but is not?
1161 Log(EVENT_LOG,LOG_NORMAL, tpparam.name,"stop_receiver_thread() - Receiver thread <" << thread_id << "> signaled for termination");
1162
1163 // signal thread for its termination
1164 recv_thread_argp->sig_terminate= true;
1165 // wait for thread to join after termination
1166 pthread_join(thread_id, 0);
1167 // the dying thread will signal main loop to call this method, but this time we should enter the else branch
1168 return;
1169 }
1170 }
1171 else
1172 Log(ERROR_LOG,LOG_NORMAL, tpparam.name,"stop_receiver_thread() - Receiver thread <" << thread_id << "> not found");
1173
1174}
1175
1176
1177/** cleans up left over structures (assoc,receiver_thread_arg) from already terminated receiver_thread
1178 * usually called by the master_thread after the receiver_thread terminated
1179 * @note clean_up_receiver_thread() should be only called when an outer lock ensures that peer_assoc
1180 * is still valid
1181 */
1182void
1183TPoverTCP::cleanup_receiver_thread(AssocData* peer_assoc)
1184{
1185 // All operations on recv_thread_argmap and connmap require an already acquired lock
1186 // after this procedure peer_assoc may be invalid because it was erased
1187
1188 // start critical section
1189
1190 if (peer_assoc == 0)
1191 return;
1192
1193 pthread_t thread_id= peer_assoc->thread_ID;
1194
1195 // try to clean up receiver_thread_arg
1196 recv_thread_argmap_t::iterator recv_thread_arg_iter= recv_thread_argmap.find(thread_id);
1197 receiver_thread_arg_t* recv_thread_argp=
1198 (recv_thread_arg_iter != recv_thread_argmap.end()) ? recv_thread_arg_iter->second : 0;
1199 if (recv_thread_argp)
1200 {
1201 if (!recv_thread_argp->terminated)
1202 {
1203 // thread signaled termination, but is not?
1204 Log(ERROR_LOG,LOG_NORMAL, tpparam.name,"cleanup_receiver_thread() - Receiver thread <" << thread_id << "> not terminated yet?!");
1205 return;
1206 }
1207 else
1208 { // if thread is already terminated
1209 Log(EVENT_LOG,LOG_NORMAL, tpparam.name,"cleanup_receiver_thread() - Receiver thread <" << thread_id << "> is terminated");
1210
1211 // delete it from receiver map
1212 recv_thread_argmap.erase(recv_thread_arg_iter);
1213
1214 // then delete receiver arg structure
1215 delete recv_thread_argp;
1216 }
1217 }
1218
1219 // delete entry from connection map
1220
1221 // cleanup sender thread
1222 // no need to lock explicitly, because caller of cleanup_receiver_thread() must already locked
1223 terminate_sender_thread(peer_assoc);
1224
1225 // delete the AssocData structure from the connection map
1226 // also frees allocated AssocData structure
1227 connmap.erase(peer_assoc);
1228
1229 // end critical section
1230
1231 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name,"cleanup_receiver_thread() - Cleanup receiver thread <" << thread_id << ">. Done.");
1232}
1233
1234
1235/* sends a stop message to the sender thread that belongs to the peer address given in assoc
1236 * @note terminate_receiver_thread() should be only called when an outer lock ensures that assoc
1237 * is still valid, a lock is also required, because senderthread_queuemap is changed
1238 */
1239void
1240TPoverTCP::terminate_sender_thread(const AssocData* assoc)
1241{
1242 if (assoc == 0)
1243 {
1244 Log(ERROR_LOG,LOG_NORMAL,tpparam.name,"terminate_sender_thread() - assoc data == NULL");
1245 return;
1246 }
1247
1248 sender_thread_queuemap_t::iterator it= senderthread_queuemap.find(assoc->peer);
1249
1250 if (it != senderthread_queuemap.end())
1251 { // we have a sender thread: send a stop message to it
1252 FastQueue* destqueue= it->second;
1253 if (destqueue)
1254 {
1255 TPoverTCPMsg* internalmsg= new TPoverTCPMsg(assoc,tpparam.source,TPoverTCPMsg::stop);
1256 if (internalmsg)
1257 {
1258 // send the internal message to the sender thread queue
1259 internalmsg->send(tpparam.source,destqueue);
1260 }
1261 }
1262 else
1263 {
1264 Log(WARNING_LOG,LOG_NORMAL,tpparam.name,"terminate_sender_thread() - found entry for address, but no sender thread. addr:" << assoc->peer);
1265 }
1266 // erase entry from map
1267 senderthread_queuemap.erase(it);
1268 }
1269}
1270
1271/* terminate all active threads
1272 * note: locking should not be necessary here because this message is called as last method from
1273 * main_loop()
1274 */
1275void
1276TPoverTCP::terminate_all_threads()
1277{
1278 AssocData* assoc= 0;
1279 receiver_thread_arg_t* terminate_argp;
1280
1281 for (recv_thread_argmap_t::iterator terminate_iterator= recv_thread_argmap.begin();
1282 terminate_iterator != recv_thread_argmap.end();
1283 terminate_iterator++)
1284 {
1285 if ( (terminate_argp= terminate_iterator->second) != 0)
1286 {
1287 // we need a non const pointer to erase it later on
1288 assoc= const_cast<AssocData*>(terminate_argp->peer_assoc);
1289 // check whether thread is still alive
1290 if (terminate_argp->terminated == false)
1291 {
1292 terminate_argp->sig_terminate= true;
1293 // then wait for its termination
1294 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name,
1295 "Signaled receiver thread <" << terminate_iterator->first << "> for termination");
1296
1297 pthread_join(terminate_iterator->first, 0);
1298
1299 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name, "Thread <" << terminate_iterator->first << "> is terminated");
1300 }
1301 else
1302 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name,
1303 "Receiver thread <" << terminate_iterator->first << "> already terminated");
1304
1305 // cleanup all remaining argument structures of terminated threads
1306 delete terminate_argp;
1307
1308 // terminate any related sender thread that is still running
1309 terminate_sender_thread(assoc);
1310
1311 connmap.erase(assoc);
1312 // delete assoc is not necessary, because connmap.erase() will do the job
1313 }
1314 } // end for
1315}
1316
1317
1318/**
1319 * sender thread starter:
1320 * just a static starter method to allow starting the
1321 * actual sender_thread() method.
1322 *
1323 * @param argp - pointer to the current TPoverTCP object instance and receiver_thread_arg_t struct
1324 */
1325void*
1326TPoverTCP::sender_thread_starter(void *argp)
1327{
1328 sender_thread_start_arg_t *sargp= static_cast<sender_thread_start_arg_t *>(argp);
1329
1330 //cout << "invoked sender_thread_Starter" << endl;
1331
1332 // invoke sender thread method
1333 if (sargp != 0 && sargp->instance != 0)
1334 {
1335 // call receiver_thread member function on object instance
1336 sargp->instance->sender_thread(sargp->sender_thread_queue);
1337
1338 //cout << "Before deletion of sarg" << endl;
1339
1340 // no longer needed
1341 delete sargp;
1342 }
1343 else
1344 {
1345 Log(ERROR_LOG,LOG_CRIT,"sender_thread_starter","while starting sender_thread: 0 pointer to arg or object");
1346 }
1347 return 0;
1348}
1349
1350
1351
1352
1353/**
1354 * receiver thread starter:
1355 * just a static starter method to allow starting the
1356 * actual receiver_thread() method.
1357 *
1358 * @param argp - pointer to the current TPoverTCP object instance and receiver_thread_arg_t struct
1359 */
1360void*
1361TPoverTCP::receiver_thread_starter(void *argp)
1362{
1363 receiver_thread_start_arg_t *rargp= static_cast<receiver_thread_start_arg_t *>(argp);
1364 // invoke receiver thread method
1365 if (rargp != 0 && rargp->instance != 0)
1366 {
1367 // call receiver_thread member function on object instance
1368 rargp->instance->receiver_thread(rargp->rtargp);
1369
1370 // no longer needed
1371 delete rargp;
1372 }
1373 else
1374 {
1375 Log(ERROR_LOG,LOG_CRIT,"receiver_thread_starter","while starting receiver_thread: 0 pointer to arg or object");
1376 }
1377 return 0;
1378}
1379
1380
1381void
1382TPoverTCP::create_new_sender_thread(FastQueue* senderfqueue)
1383{
1384 Log(EVENT_LOG,LOG_NORMAL, tpparam.name, "Starting new sender thread...");
1385
1386 pthread_t senderthreadid;
1387 // create new thread; (arg == 0) is handled by thread, too
1388 int pthread_status= pthread_create(&senderthreadid,
1389 NULL, // NULL: default attributes: thread is joinable and has a
1390 // default, non-realtime scheduling policy
1391 TPoverTCP::sender_thread_starter,
1392 new sender_thread_start_arg_t(this,senderfqueue));
1393 if (pthread_status)
1394 {
1395 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "A new thread could not be created: " << strerror(pthread_status));
1396
1397 delete senderfqueue;
1398 }
1399}
1400
1401
1402void
1403TPoverTCP::create_new_receiver_thread(AssocData* peer_assoc)
1404{
1405 receiver_thread_arg_t* argp=
1406 new(nothrow) receiver_thread_arg(peer_assoc);
1407
1408 Log(EVENT_LOG,LOG_NORMAL, tpparam.name, "Starting new receiver thread...");
1409
1410 // create new thread; (arg == 0) is handled by thread, too
1411 int pthread_status= pthread_create(&peer_assoc->thread_ID,
1412 NULL, // NULL: default attributes: thread is joinable and has a
1413 // default, non-realtime scheduling policy
1414 receiver_thread_starter,
1415 new(nothrow) receiver_thread_start_arg_t(this,argp));
1416 if (pthread_status)
1417 {
1418 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "A new thread could not be created: " << strerror(pthread_status));
1419
1420 delete argp;
1421 }
1422 else
1423 {
1424 lock(); // install_cleanup_thread_lock(TPoverTCP, this);
1425
1426 // remember pointer to thread arg structure
1427 // thread arg structure should be destroyed after thread termination only
1428 pair<recv_thread_argmap_t::iterator, bool> tmpinsiterator=
1429 recv_thread_argmap.insert( pair<pthread_t,receiver_thread_arg_t*> (peer_assoc->thread_ID,argp) );
1430 if (tmpinsiterator.second == false)
1431 {
1432 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Thread argument could not be inserted into hashmap");
1433 }
1434 unlock(); // uninstall_cleanup(1);
1435 }
1436}
1437
1438
1439/**
1440 * master listener thread starter:
1441 * just a static starter method to allow starting the
1442 * actual master_listener_thread() method.
1443 *
1444 * @param argp - pointer to the current TPoverTCP object instance
1445 */
1446void*
1447TPoverTCP::master_listener_thread_starter(void *argp)
1448{
1449 // invoke listener thread method
1450 if (argp != 0)
1451 {
1452 (static_cast<TPoverTCP*>(argp))->master_listener_thread();
1453 }
1454 return 0;
1455}
1456
1457
1458
1459/**
1460 * master listener thread: waits for incoming connections at the well-known tcp port
1461 * when a connection request is received this thread spawns a receiver_thread for
1462 * receiving packets from the peer at the new socket.
1463 */
1464void
1465TPoverTCP::master_listener_thread()
1466{
1467 // create a new address-structure for the listening masterthread
1468 struct sockaddr_in6 own_address_v6;
1469 own_address_v6.sin6_family = AF_INET6;
1470 own_address_v6.sin6_flowinfo= 0;
1471 own_address_v6.sin6_port = htons(tpparam.port); // use port number in param structure
1472 // accept incoming connections on all interfaces
1473 own_address_v6.sin6_addr = in6addr_any;
1474 own_address_v6.sin6_scope_id= 0;
1475
1476 // create a new address-structure for the listening masterthread
1477 struct sockaddr_in own_address_v4;
1478 own_address_v4.sin_family = AF_INET;
1479 own_address_v4.sin_port = htons(tpparam.port); // use port number in param structure
1480 // accept incoming connections on all interfaces
1481 own_address_v4.sin_addr.s_addr = INADDR_ANY;
1482
1483 // create a listening socket
1484 int master_listener_socket= socket( AF_INET6, SOCK_STREAM, IPPROTO_TCP);
1485 if (master_listener_socket!=-1) v4_mode = false;
1486 if (master_listener_socket == -1) {
1487 master_listener_socket= socket( AF_INET, SOCK_STREAM, IPPROTO_TCP);
1488 if (master_listener_socket!=-1) v4_mode = true;
1489 }
1490 if (master_listener_socket == -1)
1491 {
1492 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Could not create a new socket, error: " << strerror(errno));
1493 return;
1494 }
1495
1496 // Disable Nagle Algorithm, set (TCP_NODELAY)
1497 int nodelayflag= 1;
1498 int status= setsockopt(master_listener_socket,
1499 IPPROTO_TCP,
1500 TCP_NODELAY,
1501 &nodelayflag,
1502 sizeof(nodelayflag));
1503 if (status)
1504 {
1505 Log(ERROR_LOG,LOG_NORMAL,tpparam.name, "Could not set socket option TCP_NODELAY:" << strerror(errno));
1506 }
1507
1508 // Reuse ports, even if they are busy
1509 int socketreuseflag= 1;
1510 status= setsockopt(master_listener_socket,
1511 SOL_SOCKET,
1512 SO_REUSEADDR,
1513 (const char *) &socketreuseflag,
1514 sizeof(socketreuseflag));
1515 if (status)
1516 {
1517 Log(ERROR_LOG,LOG_NORMAL,tpparam.name, "Could not set socket option SO_REUSEADDR:" << strerror(errno));
1518 }
1519
1520
1521 // bind the newly created socket to a specific address
1522 int bind_status = bind(master_listener_socket, v4_mode ?
1523 reinterpret_cast<struct sockaddr *>(&own_address_v4) :
1524 reinterpret_cast<struct sockaddr *>(&own_address_v6),
1525 v4_mode ? sizeof(own_address_v4) : sizeof(own_address_v6));
1526 if (bind_status)
1527 {
1528 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Binding to "
1529 << inet_ntop(AF_INET6, &own_address_v6.sin6_addr, in6_addrstr, INET6_ADDRSTRLEN)
1530 << " port " << tpparam.port << " failed, error: " << strerror(errno));
1531 return;
1532 }
1533
1534 // listen at the socket,
1535 // queuesize for pending connections= max_listen_queue_size
1536 int listen_status = listen(master_listener_socket, max_listen_queue_size);
1537 if (listen_status)
1538 {
1539 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Listen at socket " << master_listener_socket
1540 << " failed, error: " << strerror(errno));
1541 return;
1542 }
1543 else
1544 {
1545 Log(INFO_LOG,LOG_NORMAL, tpparam.name, color[green] << "Listening at port #" << tpparam.port << color[off]);
1546 }
1547
1548 // activate O_NON_BLOCK for accept (accept does not block)
1549 fcntl(master_listener_socket,F_SETFL, O_NONBLOCK);
1550
1551 // create a pollfd struct for use in the mainloop
1552 struct pollfd poll_fd;
1553 poll_fd.fd = master_listener_socket;
1554 poll_fd.events = POLLIN | POLLPRI;
1555 poll_fd.revents = 0;
1556 /*
1557 #define POLLIN 0x001 // There is data to read.
1558 #define POLLPRI 0x002 // There is urgent data to read.
1559 #define POLLOUT 0x004 // Writing now will not block.
1560 */
1561
1562 bool terminate = false;
1563 // check for thread terminate condition using get_state()
1564 state_t currstate= get_state();
1565 int poll_status= 0;
1566 const unsigned int number_poll_sockets= 1;
1567 struct sockaddr_in6 peer_address;
1568 socklen_t peer_address_len;
1569 int conn_socket;
1570
1571 // check whether this thread is signaled for termination
1572 while(! (terminate= (currstate==STATE_ABORT || currstate==STATE_STOP) ) )
1573 {
1574 // wait on number_poll_sockets (main drm socket)
1575 // for the events specified above for sleep_time (in ms)
1576 poll_status= poll(&poll_fd, number_poll_sockets, tpparam.sleep_time);
1577 if (poll_fd.revents & POLLERR) // Error condition
1578 {
1579 if (errno != EINTR)
1580 {
1581 Log(ERROR_LOG,LOG_CRIT, tpparam.name,
1582 "Poll caused error " << strerror(errno) << " - indicated by revents");
1583 }
1584 else
1585 {
1586 Log(EVENT_LOG,LOG_NORMAL, tpparam.name, "poll(): " << strerror(errno));
1587 }
1588
1589 }
1590 if (poll_fd.revents & POLLHUP) // Hung up
1591 {
1592 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Poll hung up");
1593 return;
1594 }
1595 if (poll_fd.revents & POLLNVAL) // Invalid request: fd not open
1596 {
1597 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Poll Invalid request: fd not open");
1598 return;
1599 }
1600
1601 switch (poll_status)
1602 {
1603 case -1:
1604 if (errno != EINTR)
1605 {
1606 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Poll status indicates error: " << strerror(errno));
1607 }
1608 else
1609 {
1610 Log(EVENT_LOG,LOG_NORMAL, tpparam.name, "Poll status: " << strerror(errno));
1611 }
1612
1613 break;
1614
1615 case 0:
1616#ifdef DEBUG_HARD
1617 Log(DEBUG_LOG,LOG_UNIMP, tpparam.name,
1618 "Listen Thread - Poll timed out after " << tpparam.sleep_time << " ms.");
1619#endif
1620 currstate= get_state();
1621 continue;
1622 break;
1623
1624 default:
1625#ifdef DEBUG_HARD
1626 Log(DEBUG_LOG,LOG_UNIMP, tpparam.name, "Poll: " << poll_status << " event(s) ocurred, of type " << poll_fd.revents);
1627#endif
1628 break;
1629 } // end switch
1630
1631 // after a successful accept call,
1632 // accept stores the address information of the connecting party
1633 // in peer_address and the size of its address in addrlen
1634 peer_address_len= sizeof(peer_address);
1635 conn_socket = accept (master_listener_socket,
1636 reinterpret_cast<struct sockaddr *>(&peer_address),
1637 &peer_address_len);
1638 if (conn_socket == -1)
1639 {
1640 if (errno != EWOULDBLOCK && errno != EAGAIN)
1641 {
1642 Log(ERROR_LOG,LOG_EMERG, tpparam.name, "Accept at socket " << master_listener_socket
1643 << " failed, error: " << strerror(errno));
1644 return;
1645 }
1646 }
1647 else
1648 {
1649 // create a new assocdata-object for the new thread
1650 AssocData* peer_assoc = NULL;
1651 appladdress addr(peer_address, IPPROTO_TCP);
1652
1653 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name, "<<--Received connect--<< request from " << addr.get_ip_str()
1654 << " port #" << addr.get_port());
1655
1656 struct sockaddr_in6 own_address;
1657 if (v4_mode) {
1658 struct sockaddr_in own_address_v4;
1659 socklen_t own_address_len_v4 = sizeof(own_address_v4);
1660 getsockname(conn_socket, reinterpret_cast<struct sockaddr*>(&own_address_v4), &own_address_len_v4);
1661 v4_to_v6(&own_address_v4, &own_address);
1662 } else {
1663 socklen_t own_address_len= sizeof(own_address);
1664 getsockname(conn_socket, reinterpret_cast<struct sockaddr*>(&own_address), &own_address_len);
1665 }
1666
1667 // AssocData will copy addr content into its own structure
1668 // allocated peer_assoc will be stored in connmap
1669 peer_assoc = new(nothrow) AssocData(conn_socket, addr, appladdress(own_address,IPPROTO_TCP));
1670
1671 bool insert_success= false;
1672 if (peer_assoc)
1673 {
1674 // start critical section
1675 lock(); // install_cleanup_thread_lock(TPoverTCP, this);
1676 insert_success= connmap.insert(peer_assoc);
1677 // end critical section
1678 unlock(); // uninstall_cleanup(1);
1679 }
1680
1681
1682 if (insert_success == false) // not inserted into connmap
1683 {
1684 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Cannot insert AssocData for socket " << conn_socket
1685 << ", " << addr.get_ip_str() << ", port #"
1686 << addr.get_port() << " into connection map, aborting connection...");
1687
1688 // abort connection, delete its AssocData
1689 close (conn_socket);
1690 if (peer_assoc)
1691 {
1692 delete peer_assoc;
1693 peer_assoc= 0;
1694 }
1695 return;
1696
1697 } //end __else(connmap.insert());__
1698
1699 // create a new thread for each new connection
1700 create_new_receiver_thread(peer_assoc);
1701 } // end __else (connsocket)__
1702
1703 // get new thread state
1704 currstate= get_state();
1705
1706 } // end while(!terminate)
1707
1708 // close the listener socket
1709 close (master_listener_socket);
1710 return;
1711} // end listen_for_connections()
1712
1713
1714TPoverTCP::~TPoverTCP()
1715{
1716 init= false;
1717 this->connmap.clear();
1718
1719 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name, "Destructor called");
1720
1721 QueueManager::instance()->unregister_queue(tpparam.source);
1722}
1723
1724/** TPoverTCP Thread main loop.
1725 * This loop checks for internal messages of either
1726 * a send() call to start a new receiver thread, or,
1727 * of a receiver_thread() that signals its own termination
1728 * for proper cleanup of control structures.
1729 * It also handles the following internal TPoverTCPMsg types:
1730 * - TPoverTCPMsg::stop - a particular receiver thread is terminated
1731 * - TPoverTCPMsg::start - a particular receiver thread is started
1732 * @param nr number of current thread instance
1733 */
1734void
1735TPoverTCP::main_loop(uint32 nr)
1736{
1737
1738 // get internal queue for messages from receiver_thread
1739 FastQueue* fq = get_fqueue();
1740 if (!fq)
1741 {
1742 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Cannot find message queue");
1743 return;
1744 } // end if not fq
1745 // register queue for receiving internal messages from other modules
1746 QueueManager::instance()->register_queue(fq,tpparam.source);
1747
1748 // start master listener thread
1749 pthread_t master_listener_thread_ID;
1750 int pthread_status= pthread_create(&master_listener_thread_ID,
1751 NULL, // NULL: default attributes: thread is joinable and has a
1752 // default, non-realtime scheduling policy
1753 master_listener_thread_starter,
1754 this);
1755 if (pthread_status)
1756 {
1757 Log(ERROR_LOG,LOG_CRIT, tpparam.name,
1758 "New master listener thread could not be created: " << strerror(pthread_status));
1759 }
1760 else
1761 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name, "Master listener thread started");
1762
1763
1764 // define max latency for thread reaction on termination/stop signal
1765 timespec wait_interval= { 0, 250000000L }; // 250ms
1766 message* internal_thread_msg = NULL;
1767 state_t currstate= get_state();
1768
1769 // check whether this thread is signaled for termination
1770 while( currstate!=STATE_ABORT && currstate!=STATE_STOP )
1771 {
1772 // poll internal message queue (blocking)
1773 if ( (internal_thread_msg= fq->dequeue_timedwait(wait_interval)) != 0 )
1774 {
1775 TPoverTCPMsg* internalmsg= dynamic_cast<TPoverTCPMsg*>(internal_thread_msg);
1776 if (internalmsg)
1777 {
1778 if (internalmsg->get_msgtype() == TPoverTCPMsg::stop)
1779 {
1780 // a receiver thread terminated and signaled for cleanup by master thread
1781 AssocData* assocd= const_cast<AssocData*>(internalmsg->get_peer_assoc());
1782 Log(DEBUG_LOG,LOG_NORMAL, tpparam.name, "Got cleanup request for thread <" << assocd->thread_ID <<'>');
1783 lock();
1784 cleanup_receiver_thread( assocd );
1785 unlock();
1786 }
1787 else
1788 if (internalmsg->get_msgtype() == TPoverTCPMsg::start)
1789 {
1790 // start a new receiver thread
1791 create_new_receiver_thread( const_cast<AssocData*>(internalmsg->get_peer_assoc()) );
1792 }
1793 else
1794 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "unexpected internal message:" << internalmsg->get_msgtype());
1795
1796 delete internalmsg;
1797 }
1798 else
1799 {
1800 Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Dynamic_cast failed - received unexpected and unknown internal message source "
1801 << internal_thread_msg->get_source());
1802 }
1803 } // endif
1804
1805 // get thread state
1806 currstate= get_state();
1807 } // end while
1808
1809 if (currstate==STATE_STOP)
1810 {
1811 // start abort actions
1812 Log(INFO_LOG,LOG_NORMAL, tpparam.name, "Asked to abort, stopping all receiver threads");
1813 } // end if stopped
1814
1815 // do not accept any more messages
1816 fq->shutdown();
1817 // terminate all receiver and sender threads that are still active
1818 terminate_all_threads();
1819}
1820
1821} // end namespace protlib
1822///@}
Note: See TracBrowser for help on using the repository browser.