[5284] | 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 |
|
---|
| 30 | extern "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 |
|
---|
| 61 | const unsigned int max_listen_queue_size= 10;
|
---|
| 62 |
|
---|
[7038] | 63 | #define IPV6_ADDR_INT32_SMP 0x0000ffff
|
---|
| 64 |
|
---|
[5284] | 65 | namespace protlib {
|
---|
| 66 |
|
---|
[7038] | 67 | void 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. */
|
---|
| 75 | void 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 |
|
---|
[5284] | 85 | using namespace log;
|
---|
| 86 |
|
---|
[9686] | 87 | /** @defgroup protlib
|
---|
| 88 | * @ingroup protlib
|
---|
[5284] | 89 | * @{
|
---|
| 90 | */
|
---|
| 91 |
|
---|
| 92 | char 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 | */
|
---|
| 104 | AssocData*
|
---|
| 105 | TPoverTCP::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
|
---|
[7038] | 151 | new_socket = socket( v4_mode ? AF_INET : AF_INET6, SOCK_STREAM, IPPROTO_TCP);
|
---|
[5284] | 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 |
|
---|
[7038] | 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 | }
|
---|
[5284] | 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] << "]");
|
---|
[10074] | 207 |
|
---|
| 208 | close(new_socket);
|
---|
[5284] | 209 | return 0; // error: couldn't connect
|
---|
| 210 | }
|
---|
| 211 |
|
---|
[7038] | 212 |
|
---|
[5284] | 213 | struct sockaddr_in6 own_address;
|
---|
[7040] | 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 | }
|
---|
[5284] | 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 | {
|
---|
[8606] | 256 | bool ret = newmsg->send_to(tpparam.source);
|
---|
| 257 | if(!ret) delete newmsg;
|
---|
[5284] | 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 | */
|
---|
| 291 | void
|
---|
| 292 | TPoverTCP::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 | */
|
---|
| 357 | void
|
---|
| 358 | TPoverTCP::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 |
|
---|
[8606] | 417 | appladdress* apl=new appladdress(*addr);
|
---|
| 418 | TPoverTCPMsg* internalmsg= new TPoverTCPMsg(netmsg,apl);
|
---|
[5284] | 419 | if (internalmsg)
|
---|
| 420 | {
|
---|
| 421 | // send the internal message to the sender thread queue
|
---|
[8606] | 422 | bool sent = internalmsg->send(tpparam.source,destqueue);
|
---|
| 423 | if (!sent) {
|
---|
| 424 | delete internalmsg->get_appladdr();
|
---|
| 425 | delete internalmsg;
|
---|
| 426 | internalmsg = NULL;
|
---|
| 427 | }
|
---|
[5284] | 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 | */
|
---|
| 454 | void
|
---|
| 455 | TPoverTCP::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 | */
|
---|
| 643 | void
|
---|
| 644 | TPoverTCP::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 | {
|
---|
[8606] | 669 | ERRLog(tpparam.name, methodname << "received not an TPoverTCPMsg but a" << internal_thread_msg->get_type_name());
|
---|
[5284] | 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;
|
---|
[8606] | 704 | }
|
---|
| 705 |
|
---|
| 706 | delete internalmsg;
|
---|
[5284] | 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 | */
|
---|
| 720 | void
|
---|
| 721 | TPoverTCP::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 | {
|
---|
[8606] | 941 | delete netmsg;
|
---|
[5284] | 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;
|
---|
[8609] | 1119 | //delete netmsg;
|
---|
[5284] | 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 | */
|
---|
| 1139 | void
|
---|
| 1140 | TPoverTCP::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 | */
|
---|
| 1182 | void
|
---|
| 1183 | TPoverTCP::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 | */
|
---|
| 1239 | void
|
---|
| 1240 | TPoverTCP::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 | */
|
---|
| 1275 | void
|
---|
| 1276 | TPoverTCP::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 | */
|
---|
| 1325 | void*
|
---|
| 1326 | TPoverTCP::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 | */
|
---|
| 1360 | void*
|
---|
| 1361 | TPoverTCP::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 |
|
---|
| 1381 | void
|
---|
| 1382 | TPoverTCP::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 |
|
---|
| 1402 | void
|
---|
| 1403 | TPoverTCP::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 | */
|
---|
| 1446 | void*
|
---|
| 1447 | TPoverTCP::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 | */
|
---|
| 1464 | void
|
---|
| 1465 | TPoverTCP::master_listener_thread()
|
---|
| 1466 | {
|
---|
| 1467 | // create a new address-structure for the listening masterthread
|
---|
[7038] | 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
|
---|
[5284] | 1472 | // accept incoming connections on all interfaces
|
---|
[7038] | 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;
|
---|
[5284] | 1482 |
|
---|
| 1483 | // create a listening socket
|
---|
[7038] | 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 | }
|
---|
[5284] | 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
|
---|
[7038] | 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));
|
---|
[5284] | 1526 | if (bind_status)
|
---|
| 1527 | {
|
---|
| 1528 | Log(ERROR_LOG,LOG_CRIT, tpparam.name, "Binding to "
|
---|
[10073] | 1529 | << inet_ntop(AF_INET6, &own_address_v6.sin6_addr, in6_addrstr, INET6_ADDRSTRLEN)
|
---|
[5284] | 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;
|
---|
[7038] | 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 | }
|
---|
[5284] | 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)
|
---|
[10432] | 1707 |
|
---|
| 1708 | // close the listener socket
|
---|
| 1709 | close (master_listener_socket);
|
---|
[5284] | 1710 | return;
|
---|
| 1711 | } // end listen_for_connections()
|
---|
| 1712 |
|
---|
| 1713 |
|
---|
| 1714 | TPoverTCP::~TPoverTCP()
|
---|
| 1715 | {
|
---|
| 1716 | init= false;
|
---|
[8606] | 1717 | this->connmap.clear();
|
---|
[5284] | 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 | */
|
---|
| 1734 | void
|
---|
| 1735 | TPoverTCP::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 | ///@}
|
---|