From f4ac8d7a443bd1afb35634d1cccb79943cc316e0 Mon Sep 17 00:00:00 2001 From: Martin Sustrik Date: Sun, 13 Dec 2009 09:56:02 +0100 Subject: OpenPGM v1 support removed --- src/pgm_socket.cpp | 226 ++--------------------------------------------------- 1 file changed, 5 insertions(+), 221 deletions(-) (limited to 'src/pgm_socket.cpp') diff --git a/src/pgm_socket.cpp b/src/pgm_socket.cpp index 8400b83..60712f4 100644 --- a/src/pgm_socket.cpp +++ b/src/pgm_socket.cpp @@ -127,13 +127,9 @@ int zmq::pgm_socket_t::open_transport (void) nbytes_processed = 0; pgm_msgv_processed = 0; -#ifdef ZMQ_HAVE_OPENPGM1 - int pgm_ok = 0; -#elif defined ZMQ_HAVE_OPENPGM2 + // TODO: Converting bool to int? Not nice. int pgm_ok = true; - GError *pgm_error = NULL; -#endif // Init PGM transport. // Ensure threading enabled, ensure timer enabled and find PGM protocol id. @@ -162,11 +158,7 @@ int zmq::pgm_socket_t::open_transport (void) gsi_base = uuid_t ().to_string (); } -#ifdef ZMQ_HAVE_OPENPGM1 - rc = pgm_create_custom_gsi (gsi_base.c_str (), &gsi); -#elif defined ZMQ_HAVE_OPENPGM2 - rc = pgm_gsi_create_from_string (&gsi, gsi_base.c_str (), -1); -#endif + rc = pgm_gsi_create_from_string (&gsi, gsi_base.c_str (), -1); if (rc != pgm_ok) { errno = EINVAL; @@ -176,26 +168,6 @@ int zmq::pgm_socket_t::open_transport (void) //zmq_log (1, "Transport GSI: %s, %s(%i)\n", pgm_print_gsi (&gsi), // __FILE__, __LINE__); -#ifdef ZMQ_HAVE_OPENPGM1 - // PGM transport GSRs. - struct group_source_req recv_gsr, send_gsr; - size_t recv_gsr_len = 1; - - // On success, 0 is returned. On invalid arguments, -EINVAL is returned. - // If more multicast groups are found than the recv_len parameter, - // -ENOMEM is returned. - rc = pgm_if_parse_transport (network, AF_INET, &recv_gsr, - &recv_gsr_len, &send_gsr); - if (rc != 0) { - errno = EINVAL; - return -1; - } - - if (recv_gsr_len != 1) { - errno = ENOMEM; - return -1; - } -#elif defined ZMQ_HAVE_OPENPGM2 struct pgm_transport_info_t *res = NULL; if (!pgm_if_get_transport_info (network, NULL, &res, &pgm_error)) { @@ -205,29 +177,13 @@ int zmq::pgm_socket_t::open_transport (void) res->ti_gsi = gsi; res->ti_dport = port_number; -#endif // If we are using UDP encapsulation update gsr or res. if (udp_encapsulation) { -#ifdef ZMQ_HAVE_OPENPGM1 - // Use the same port for UDP encapsulation. - ((struct sockaddr_in*)&send_gsr.gsr_group)->sin_port = - g_htons (port_number); - ((struct sockaddr_in*)&recv_gsr.gsr_group)->sin_port = - g_htons (port_number); -#elif defined ZMQ_HAVE_OPENPGM2 res->ti_udp_encap_ucast_port = port_number; res->ti_udp_encap_mcast_port = port_number; -#endif } -#ifdef ZMQ_HAVE_OPENPGM1 - rc = pgm_transport_create (&transport, &gsi, 0, port_number, &recv_gsr, - 1, &send_gsr); - if (rc != 0) { - return -1; - } -#elif defined ZMQ_HAVE_OPENPGM2 if (!pgm_transport_create (&transport, res, &pgm_error)) { pgm_if_free_transport_info (res); // TODO: tranlate errors from glib into errnos. @@ -236,7 +192,6 @@ int zmq::pgm_socket_t::open_transport (void) } pgm_if_free_transport_info (res); -#endif zmq_log (1, "PGM transport created, %s(%i)\n", __FILE__, __LINE__); @@ -270,11 +225,8 @@ int zmq::pgm_socket_t::open_transport (void) // Set transport->can_send_data = FALSE. // Note that NAKs are still generated by the transport. -#if defined ZMQ_HAVE_OPENPGM1 - rc = pgm_transport_set_recv_only (transport, false); -#elif defined ZMQ_HAVE_OPENPGM2 rc = pgm_transport_set_recv_only (transport, true, false); -#endif + zmq_assert (rc == pgm_ok); // Set NAK transmit back-off interval [us]. @@ -361,23 +313,6 @@ int zmq::pgm_socket_t::open_transport (void) return -1; } -#ifdef ZMQ_HAVE_OPENPGM1 - // Preallocate full transmit window. For simplification always - // worst case is used (40 bytes ipv6 header and 20 bytes UDP - // encapsulation). - int to_preallocate = options.recovery_ivl * (options.rate * 1000 / 8) - / (pgm_max_tpdu - 40 - 20); - - rc = pgm_transport_set_txw_preallocate (transport, to_preallocate); - if (rc != 0) { - errno = EINVAL; - return -1; - } - - zmq_log (2, "Preallocated %i slices in TX window. %s(%i)\n", - to_preallocate, __FILE__, __LINE__); -#endif - // Set interval of background SPM packets [us]. rc = pgm_transport_set_ambient_spm (transport, 8192 * 1000); zmq_assert (rc == pgm_ok); @@ -398,17 +333,10 @@ int zmq::pgm_socket_t::open_transport (void) } // Bind a transport to the specified network devices. -#ifdef ZMQ_HAVE_OPENPGM1 - rc = pgm_transport_bind (transport); - if (rc != 0) { - return -1; - } -#elif defined ZMQ_HAVE_OPENPGM2 - if (!pgm_transport_bind (transport, &pgm_error)) { + if (!pgm_transport_bind (transport, &pgm_error)) { // TODO: tranlate errors from glib into errnos. return -1; - } -#endif + } zmq_log (1, "PGM transport bound, %s(%i)\n", __FILE__, __LINE__); @@ -444,28 +372,6 @@ int zmq::pgm_socket_t::get_receiver_fds (int *receive_fd_, zmq_assert (receive_fd_); zmq_assert (waiting_pipe_fd_); -#if defined ZMQ_HAVE_OPENPGM1 - // For POLLIN there are 2 pollfds in pgm_transport. - int fds_array_size = pgm_receiver_fd_count; - pollfd *fds = new pollfd [fds_array_size]; - memset (fds, '\0', fds_array_size * sizeof (fds)); - - // Retrieve pollfds from pgm_transport. - int rc = pgm_transport_poll_info (transport, fds, &fds_array_size, - POLLIN); - - // pgm_transport_poll_info has to return 2 pollfds for POLLIN. - // Note that fds_array_size parameter can be - // changed inside pgm_transport_poll_info call. - zmq_assert (rc == pgm_receiver_fd_count); - - // Store pfds into user allocated space. - *receive_fd_ = fds [0].fd; - *waiting_pipe_fd_ = fds [1].fd; - - delete [] fds; - -#elif defined ZMQ_HAVE_OPENPGM2 // recv_sock2 should not be used - check it. zmq_assert (transport->recv_sock2 == -1); @@ -476,7 +382,6 @@ int zmq::pgm_socket_t::get_receiver_fds (int *receive_fd_, // Take FDs directly from transport. *receive_fd_ = pgm_transport_get_recv_fd (transport); *waiting_pipe_fd_ = pgm_transport_get_pending_fd (transport); -#endif return pgm_receiver_fd_count; } @@ -491,31 +396,6 @@ int zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_, zmq_assert (send_fd_); zmq_assert (receive_fd_); -#if defined ZMQ_HAVE_OPENPGM1 - zmq_assert (!rdata_notify_fd_); - - // Preallocate pollfds array. - int fds_array_size = pgm_sender_fd_count; - pollfd *fds = new pollfd [fds_array_size]; - memset (fds, '\0', fds_array_size * sizeof (fds)); - - // Retrieve pollfds from pgm_transport. - int rc = pgm_transport_poll_info (transport, fds, &fds_array_size, - POLLOUT | POLLIN); - - // pgm_transport_poll_info has to return one pollfds for POLLOUT and - // second for POLLIN. - // Note that fds_array_size parameter can be - // changed inside pgm_transport_poll_info call. - zmq_assert (rc == pgm_sender_fd_count); - - // Store pfds into user allocated space. - *receive_fd_ = fds [0].fd; - *send_fd_ = fds [1].fd; - - delete [] fds; - -#elif defined ZMQ_HAVE_OPENPGM2 zmq_assert (rdata_notify_fd_); // recv_sock2 should not be used - check it. @@ -529,7 +409,6 @@ int zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_, *receive_fd_ = pgm_transport_get_recv_fd (transport); *rdata_notify_fd_ = pgm_transport_get_repair_fd (transport); *send_fd_ = pgm_transport_get_send_fd (transport); -#endif return pgm_sender_fd_count; } @@ -537,27 +416,6 @@ int zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_, // Send one APDU, transmit window owned memory. size_t zmq::pgm_socket_t::send (unsigned char *data_, size_t data_len_) { - -#if defined ZMQ_HAVE_OPENPGM1 - - ssize_t nbytes = 0; - iovec iov = {data_,data_len_}; - - nbytes = pgm_transport_send_packetv (transport, &iov, 1, - MSG_DONTWAIT | MSG_WAITALL, true); - - zmq_assert (nbytes != -EINVAL); - - if (nbytes == -1 && errno != EAGAIN) { - errno_assert (false); - } - - // If nbytes is -1 and errno is EAGAIN means that we can not send data - // now. We have to call write_one_pkt again. - nbytes = nbytes == -1 ? 0 : nbytes; - -#elif defined ZMQ_HAVE_OPENPGM2 - size_t nbytes = 0; PGMIOStatus status = pgm_send (transport, data_, data_len_, &nbytes); @@ -569,7 +427,6 @@ size_t zmq::pgm_socket_t::send (unsigned char *data_, size_t data_len_) zmq_assert (status == PGM_IO_STATUS_RATE_LIMITED); zmq_assert (nbytes == 0); } -#endif zmq_log (4, "wrote %i/%iB, %s(%i)\n", (int) nbytes, (int) data_len_, __FILE__, __LINE__); @@ -618,26 +475,17 @@ void *zmq::pgm_socket_t::get_buffer (size_t *size_) // Store size. *size_ = get_max_tsdu_size (); -#if defined ZMQ_HAVE_OPENPGM1 - // Allocate one packet in tx window. - return pgm_packetv_alloc (transport, false); -#elif defined ZMQ_HAVE_OPENPGM2 // Allocate buffer. unsigned char *apdu_buff = new unsigned char [*size_]; zmq_assert (apdu_buff); return apdu_buff; -#endif } // Return an unused packet allocated from the transmit window // via pgm_packetv_alloc(). void zmq::pgm_socket_t::free_buffer (void *data_) { -#if defined ZMQ_HAVE_OPENPGM1 - pgm_packetv_free1 (transport, data_, false); -#elif defined ZMQ_HAVE_OPENPGM2 delete [] (unsigned char*) data_; -#endif } @@ -672,42 +520,6 @@ ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_) // Receive a vector of Application Protocol Domain Unit's (APDUs) // from the transport. -#ifdef ZMQ_HAVE_OPENPGM1 - nbytes_rec = pgm_transport_recvmsgv (transport, pgm_msgv, - pgm_msgv_len, MSG_DONTWAIT); - - // In a case when no ODATA/RDATA fired POLLIN event (SPM...) - // pgm_transport_recvmsg returns -1 with errno == EAGAIN. - if (nbytes_rec == -1 && errno == EAGAIN) { - - // In case if no RDATA/ODATA caused POLLIN 0 is - // returned. - nbytes_rec = 0; - return 0; - } - - // For data loss nbytes_rec == -1 errno == ECONNRESET. - if (nbytes_rec == -1 && errno == ECONNRESET) { - - // Save lost data TSI. - *tsi_ = &transport->lost_data_tsi; - - zmq_log (1, "Data loss detected %s, %s(%i)\n", - pgm_print_tsi (&transport->lost_data_tsi), __FILE__, __LINE__); - - nbytes_rec = 0; - - // In case of dala loss -1 is returned. - return -1; - } - - // Catch the rest of the errors. - if (nbytes_rec <= 0) { - zmq_log (2, "received %i B, errno %i, %s(%i).\n", (int)nbytes_rec, - errno, __FILE__, __LINE__); - errno_assert (false); - } -#elif defined ZMQ_HAVE_OPENPGM2 GError *pgm_error = NULL; const PGMIOStatus status = pgm_recvmsgv (transport, pgm_msgv, @@ -760,7 +572,6 @@ ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_) nbytes_rec = 0; return -1; } -#endif zmq_log (4, "received %i bytes\n", (int)nbytes_rec); @@ -768,17 +579,6 @@ ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_) zmq_assert (nbytes_rec > 0); -#ifdef ZMQ_HAVE_OPENPGM1 - // Only one APDU per pgm_msgv_t structure is allowed. - zmq_assert (pgm_msgv [pgm_msgv_processed].msgv_iovlen == 1); - - // Take pointers from pgm_msgv_t structure. - *raw_data_ = pgm_msgv[pgm_msgv_processed].msgv_iov->iov_base; - raw_data_len = pgm_msgv[pgm_msgv_processed].msgv_iov->iov_len; - - // Save current TSI. - *tsi_ = pgm_msgv [pgm_msgv_processed].msgv_tsi; -#elif defined ZMQ_HAVE_OPENPGM2 // Only one APDU per pgm_msgv_t structure is allowed. zmq_assert (pgm_msgv [pgm_msgv_processed].msgv_len == 1); @@ -791,7 +591,6 @@ ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_) // Save current TSI. *tsi_ = &skb->tsi; -#endif // Move the the next pgm_msgv_t structure. pgm_msgv_processed++; @@ -808,19 +607,6 @@ void zmq::pgm_socket_t::process_upstream (void) pgm_msgv_t dummy_msg; -#ifdef ZMQ_HAVE_OPENPGM1 - ssize_t dummy_bytes = 0; - - // We acctually do not want to read any data here we are going to - // process NAK. - - dummy_bytes = pgm_transport_recvmsgv (transport, &dummy_msg, - 1, MSG_DONTWAIT); - - // No data should be returned. - zmq_assert (dummy_bytes == -1 && errno == EAGAIN); - -#elif defined ZMQ_HAVE_OPENPGM2 size_t dummy_bytes = 0; GError *pgm_error = NULL; @@ -833,8 +619,6 @@ void zmq::pgm_socket_t::process_upstream (void) // No data should be returned. zmq_assert (dummy_bytes == 0 && (status == PGM_IO_STATUS_TIMER_PENDING || status == PGM_IO_STATUS_RATE_LIMITED)); -#endif - } #endif -- cgit v1.2.3