From 72c5c5fff42fc0b4c9d1eaaaebe9d6e1dd8824f2 Mon Sep 17 00:00:00 2001 From: malosek Date: Thu, 24 Sep 2009 16:23:49 +0200 Subject: --with-pgm2 compilable --- src/pgm_socket.cpp | 91 +++++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 79 insertions(+), 12 deletions(-) (limited to 'src/pgm_socket.cpp') diff --git a/src/pgm_socket.cpp b/src/pgm_socket.cpp index 0d7a3e2..c35870d 100644 --- a/src/pgm_socket.cpp +++ b/src/pgm_socket.cpp @@ -19,9 +19,12 @@ #include "platform.hpp" -#if defined ZMQ_HAVE_OPENPGM1 +#ifdef ZMQ_HAVE_OPENPGM #ifdef ZMQ_HAVE_LINUX +// TODO: add this into platform.hpp? +#define CONFIG_HAVE_POLL + #include #include #endif @@ -151,10 +154,6 @@ int zmq::pgm_socket_t::open_transport (void) // PGM transport GSI. pgm_gsi_t gsi; - // PGM transport GSRs. - struct group_source_req recv_gsr, send_gsr; - size_t recv_gsr_len = 1; - if (options.identity.size () > 0) { // Create gsi from identity string. @@ -174,6 +173,11 @@ 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. @@ -188,24 +192,54 @@ int zmq::pgm_socket_t::open_transport (void) errno = ENOMEM; return -1; } +#endif - // If we are using UDP encapsulation update send_gsr & recv_gsr - // structures. Note that send_gsr & recv_gsr has to be updated after - // pgm_if_parse_transport call. - if (udp_encapsulation) { +#ifdef ZMQ_HAVE_OPENPGM2 + struct pgm_transport_info_t* res = NULL; + + if (!pgm_if_get_transport_info (network, NULL, &res, NULL)) { + errno = EINVAL; + return -1; + } + + res->ti_gsi = gsi; +#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); +#endif + +#ifdef 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 (&g_transport, &gsi, 0, port_number, &recv_gsr, 1, &send_gsr); if (rc != 0) { return -1; } +#endif + +#ifdef ZMQ_HAVE_OPENPGM2 + if (!pgm_transport_create (&g_transport, res, NULL)) { + pgm_if_free_transport_info (res); + // TODO: tranlate errors from glib into errnos. + errno = EINVAL; + return -1; + } + + pgm_if_free_transport_info (res); +#endif // Common parameters for receiver and sender. @@ -347,6 +381,7 @@ 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). @@ -361,6 +396,7 @@ int zmq::pgm_socket_t::open_transport (void) 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 (g_transport, 8192 * 1000); @@ -392,10 +428,19 @@ int zmq::pgm_socket_t::open_transport (void) } // Bind a transport to the specified network devices. +#ifdef ZMQ_HAVE_OPENPGM1 rc = pgm_transport_bind (g_transport); if (rc != 0) { return -1; } +#endif + +#ifdef ZMQ_HAVE_OPENPGM2 + if (!pgm_transport_bind (g_transport, NULL)) { + // TODO: tranlate errors from glib into errnos. + return -1; + } +#endif return 0; } @@ -484,9 +529,13 @@ int zmq::pgm_socket_t::get_sender_fds (int *send_fd_, int *receive_fd_) size_t zmq::pgm_socket_t::send (unsigned char *data_, size_t data_len_) { + ssize_t nbytes = 0; + +#ifdef ZMQ_HAVE_OPENPGM1 + iovec iov = {data_,data_len_}; - ssize_t nbytes = pgm_transport_send_packetv (g_transport, &iov, 1, + nbytes = pgm_transport_send_packetv (g_transport, &iov, 1, MSG_DONTWAIT | MSG_WAITALL, true); zmq_assert (nbytes != -EINVAL); @@ -505,6 +554,7 @@ size_t zmq::pgm_socket_t::send (unsigned char *data_, size_t data_len_) if (nbytes > 0) { zmq_assert (nbytes == (ssize_t)data_len_); } +#endif return nbytes; } @@ -535,6 +585,7 @@ size_t zmq::pgm_socket_t::get_max_apdu_at_once (size_t readbuf_size_) return apdu_count; } +#ifdef ZMQ_HAVE_OPENPGM1 // Allocate buffer for one packet from the transmit window, The memory buffer // is owned by the transmit window and so must be returned to the window with // content via pgm_transport_send() calls or unused with pgm_packetv_free1(). @@ -553,12 +604,17 @@ void zmq::pgm_socket_t::free_buffer (void *data_) { pgm_packetv_free1 (g_transport, data_, false); } +#endif // pgm_transport_recvmsgv is called to fill the pgm_msgv array up to // pgm_msgv_len. In subsequent calls data from pgm_msgv structure are // returned. ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_) { + + size_t raw_data_len = 0; + +#ifdef ZMQ_HAVE_OPENPGM1 // We just sent all data from pgm_transport_recvmsgv up // and have to return 0 that another engine in this thread is scheduled. if (nbytes_rec == nbytes_processed && nbytes_rec > 0) { @@ -626,7 +682,7 @@ ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_) // Take pointers from pgm_msgv_t structure. *raw_data_ = pgm_msgv[pgm_msgv_processed].msgv_iov->iov_base; - size_t raw_data_len = pgm_msgv[pgm_msgv_processed].msgv_iov->iov_len; + raw_data_len = pgm_msgv[pgm_msgv_processed].msgv_iov->iov_len; // Save current TSI. *tsi_ = pgm_msgv [pgm_msgv_processed].msgv_tsi; @@ -635,6 +691,8 @@ ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_) pgm_msgv_processed++; nbytes_processed +=raw_data_len; +#endif + zmq_log (4, "sendig up %i bytes\n", (int)raw_data_len); return raw_data_len; @@ -643,12 +701,21 @@ ssize_t zmq::pgm_socket_t::receive (void **raw_data_, const pgm_tsi_t **tsi_) void zmq::pgm_socket_t::process_upstream (void) { zmq_log (1, "On upstream packet, %s(%i)\n", __FILE__, __LINE__); + + ssize_t dummy_bytes = 0; + +#ifdef ZMQ_HAVE_OPENPGM1 // We acctually do not want to read any data here we are going to // process NAK. pgm_msgv_t dummy_msg; - ssize_t dummy_bytes = pgm_transport_recvmsgv (g_transport, &dummy_msg, + dummy_bytes = pgm_transport_recvmsgv (g_transport, &dummy_msg, 1, MSG_DONTWAIT); +#endif + +#ifdef ZMQ_HAVE_OPENPGM2 + zmq_assert (false); +#endif // No data should be returned. zmq_assert (dummy_bytes == -1 && errno == EAGAIN); -- cgit v1.2.3