1/*
2 Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
3
4 This file is part of libzmq, the ZeroMQ core engine in C++.
5
6 libzmq is free software; you can redistribute it and/or modify it under
7 the terms of the GNU Lesser General Public License (LGPL) as published
8 by the Free Software Foundation; either version 3 of the License, or
9 (at your option) any later version.
10
11 As a special exception, the Contributors give you permission to link
12 this library with independent modules to produce an executable,
13 regardless of the license terms of these independent modules, and to
14 copy and distribute the resulting executable under terms of your choice,
15 provided that you also meet, for each linked independent module, the
16 terms and conditions of the license of that module. An independent
17 module is a module which is not derived from or based on this library.
18 If you modify this library, you must extend this exception to your
19 version of the library.
20
21 libzmq is distributed in the hope that it will be useful, but WITHOUT
22 ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
23 FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
24 License for more details.
25
26 You should have received a copy of the GNU Lesser General Public License
27 along with this program. If not, see <http://www.gnu.org/licenses/>.
28*/
29
30#include "precompiled.hpp"
31#include <new>
32#include <string>
33
34#include "macros.hpp"
35#include "tcp_connecter.hpp"
36#include "io_thread.hpp"
37#include "err.hpp"
38#include "ip.hpp"
39#include "tcp.hpp"
40#include "address.hpp"
41#include "tcp_address.hpp"
42#include "session_base.hpp"
43
44#if !defined ZMQ_HAVE_WINDOWS
45#include <unistd.h>
46#include <sys/types.h>
47#include <sys/socket.h>
48#include <arpa/inet.h>
49#include <netinet/tcp.h>
50#include <netinet/in.h>
51#include <netdb.h>
52#include <fcntl.h>
53#ifdef ZMQ_HAVE_VXWORKS
54#include <sockLib.h>
55#endif
56#ifdef ZMQ_HAVE_OPENVMS
57#include <ioctl.h>
58#endif
59#endif
60
61#ifdef __APPLE__
62#include <TargetConditionals.h>
63#endif
64
65zmq::tcp_connecter_t::tcp_connecter_t (class io_thread_t *io_thread_,
66 class session_base_t *session_,
67 const options_t &options_,
68 address_t *addr_,
69 bool delayed_start_) :
70 stream_connecter_base_t (
71 io_thread_, session_, options_, addr_, delayed_start_),
72 _connect_timer_started (false)
73{
74 zmq_assert (_addr->protocol == protocol_name::tcp);
75}
76
77zmq::tcp_connecter_t::~tcp_connecter_t ()
78{
79 zmq_assert (!_connect_timer_started);
80}
81
82void zmq::tcp_connecter_t::process_term (int linger_)
83{
84 if (_connect_timer_started) {
85 cancel_timer (connect_timer_id);
86 _connect_timer_started = false;
87 }
88
89 stream_connecter_base_t::process_term (linger_);
90}
91
92void zmq::tcp_connecter_t::out_event ()
93{
94 if (_connect_timer_started) {
95 cancel_timer (connect_timer_id);
96 _connect_timer_started = false;
97 }
98
99 // TODO this is still very similar to (t)ipc_connecter_t, maybe the
100 // differences can be factored out
101
102 rm_handle ();
103
104 const fd_t fd = connect ();
105
106 // Handle the error condition by attempt to reconnect.
107 if (fd == retired_fd || !tune_socket (fd)) {
108 close ();
109 add_reconnect_timer ();
110 return;
111 }
112
113 create_engine (fd, get_socket_name<tcp_address_t> (fd, socket_end_local));
114}
115
116void zmq::tcp_connecter_t::timer_event (int id_)
117{
118 if (id_ == connect_timer_id) {
119 _connect_timer_started = false;
120 rm_handle ();
121 close ();
122 add_reconnect_timer ();
123 } else
124 stream_connecter_base_t::timer_event (id_);
125}
126
127void zmq::tcp_connecter_t::start_connecting ()
128{
129 // Open the connecting socket.
130 const int rc = open ();
131
132 // Connect may succeed in synchronous manner.
133 if (rc == 0) {
134 _handle = add_fd (_s);
135 out_event ();
136 }
137
138 // Connection establishment may be delayed. Poll for its completion.
139 else if (rc == -1 && errno == EINPROGRESS) {
140 _handle = add_fd (_s);
141 set_pollout (_handle);
142 _socket->event_connect_delayed (
143 make_unconnected_connect_endpoint_pair (_endpoint), zmq_errno ());
144
145 // add userspace connect timeout
146 add_connect_timer ();
147 }
148
149 // Handle any other error condition by eventual reconnect.
150 else {
151 if (_s != retired_fd)
152 close ();
153 add_reconnect_timer ();
154 }
155}
156
157void zmq::tcp_connecter_t::add_connect_timer ()
158{
159 if (options.connect_timeout > 0) {
160 add_timer (options.connect_timeout, connect_timer_id);
161 _connect_timer_started = true;
162 }
163}
164
165int zmq::tcp_connecter_t::open ()
166{
167 zmq_assert (_s == retired_fd);
168
169 // Resolve the address
170 if (_addr->resolved.tcp_addr != NULL) {
171 LIBZMQ_DELETE (_addr->resolved.tcp_addr);
172 }
173
174 _addr->resolved.tcp_addr = new (std::nothrow) tcp_address_t ();
175 alloc_assert (_addr->resolved.tcp_addr);
176 _s = tcp_open_socket (_addr->address.c_str (), options, false, true,
177 _addr->resolved.tcp_addr);
178 if (_s == retired_fd) {
179 // TODO we should emit some event in this case!
180
181 LIBZMQ_DELETE (_addr->resolved.tcp_addr);
182 return -1;
183 }
184 zmq_assert (_addr->resolved.tcp_addr != NULL);
185
186 // Set the socket to non-blocking mode so that we get async connect().
187 unblock_socket (_s);
188
189 const tcp_address_t *const tcp_addr = _addr->resolved.tcp_addr;
190
191 int rc;
192
193 // Set a source address for conversations
194 if (tcp_addr->has_src_addr ()) {
195 // Allow reusing of the address, to connect to different servers
196 // using the same source port on the client.
197 int flag = 1;
198#ifdef ZMQ_HAVE_WINDOWS
199 rc = setsockopt (_s, SOL_SOCKET, SO_REUSEADDR,
200 reinterpret_cast<const char *> (&flag), sizeof (int));
201 wsa_assert (rc != SOCKET_ERROR);
202#elif defined ZMQ_HAVE_VXWORKS
203 rc = setsockopt (_s, SOL_SOCKET, SO_REUSEADDR, (char *) &flag,
204 sizeof (int));
205 errno_assert (rc == 0);
206#else
207 rc = setsockopt (_s, SOL_SOCKET, SO_REUSEADDR, &flag, sizeof (int));
208 errno_assert (rc == 0);
209#endif
210
211#if defined ZMQ_HAVE_VXWORKS
212 rc = ::bind (_s, (sockaddr *) tcp_addr->src_addr (),
213 tcp_addr->src_addrlen ());
214#else
215 rc = ::bind (_s, tcp_addr->src_addr (), tcp_addr->src_addrlen ());
216#endif
217 if (rc == -1)
218 return -1;
219 }
220
221 // Connect to the remote peer.
222#if defined ZMQ_HAVE_VXWORKS
223 rc = ::connect (_s, (sockaddr *) tcp_addr->addr (), tcp_addr->addrlen ());
224#else
225 rc = ::connect (_s, tcp_addr->addr (), tcp_addr->addrlen ());
226#endif
227 // Connect was successful immediately.
228 if (rc == 0) {
229 return 0;
230 }
231
232 // Translate error codes indicating asynchronous connect has been
233 // launched to a uniform EINPROGRESS.
234#ifdef ZMQ_HAVE_WINDOWS
235 const int last_error = WSAGetLastError ();
236 if (last_error == WSAEINPROGRESS || last_error == WSAEWOULDBLOCK)
237 errno = EINPROGRESS;
238 else
239 errno = wsa_error_to_errno (last_error);
240#else
241 if (errno == EINTR)
242 errno = EINPROGRESS;
243#endif
244 return -1;
245}
246
247zmq::fd_t zmq::tcp_connecter_t::connect ()
248{
249 // Async connect has finished. Check whether an error occurred
250 int err = 0;
251#if defined ZMQ_HAVE_HPUX || defined ZMQ_HAVE_VXWORKS
252 int len = sizeof err;
253#else
254 socklen_t len = sizeof err;
255#endif
256
257 const int rc = getsockopt (_s, SOL_SOCKET, SO_ERROR,
258 reinterpret_cast<char *> (&err), &len);
259
260 // Assert if the error was caused by 0MQ bug.
261 // Networking problems are OK. No need to assert.
262#ifdef ZMQ_HAVE_WINDOWS
263 zmq_assert (rc == 0);
264 if (err != 0) {
265 if (err == WSAEBADF || err == WSAENOPROTOOPT || err == WSAENOTSOCK
266 || err == WSAENOBUFS) {
267 wsa_assert_no (err);
268 }
269 return retired_fd;
270 }
271#else
272 // Following code should handle both Berkeley-derived socket
273 // implementations and Solaris.
274 if (rc == -1)
275 err = errno;
276 if (err != 0) {
277 errno = err;
278#if !defined(TARGET_OS_IPHONE) || !TARGET_OS_IPHONE
279 errno_assert (errno != EBADF && errno != ENOPROTOOPT
280 && errno != ENOTSOCK && errno != ENOBUFS);
281#else
282 errno_assert (errno != ENOPROTOOPT && errno != ENOTSOCK
283 && errno != ENOBUFS);
284#endif
285 return retired_fd;
286 }
287#endif
288
289 // Return the newly connected socket.
290 const fd_t result = _s;
291 _s = retired_fd;
292 return result;
293}
294
295bool zmq::tcp_connecter_t::tune_socket (const fd_t fd_)
296{
297 const int rc = tune_tcp_socket (fd_)
298 | tune_tcp_keepalives (
299 fd_, options.tcp_keepalive, options.tcp_keepalive_cnt,
300 options.tcp_keepalive_idle, options.tcp_keepalive_intvl)
301 | tune_tcp_maxrt (fd_, options.tcp_maxrt);
302 return rc == 0;
303}
304