Kea  2.3.7
tcp_socket.h
Go to the documentation of this file.
1 // Copyright (C) 2011-2022 Internet Systems Consortium, Inc. ("ISC")
2 //
3 // This Source Code Form is subject to the terms of the Mozilla Public
4 // License, v. 2.0. If a copy of the MPL was not distributed with this
5 // file, You can obtain one at http://mozilla.org/MPL/2.0/.
6 
7 #ifndef TCP_SOCKET_H
8 #define TCP_SOCKET_H 1
9 
10 #ifndef BOOST_ASIO_HPP
11 #error "asio.hpp must be included before including this, see asiolink.h as to why"
12 #endif
13 
14 #include <netinet/in.h>
15 #include <sys/socket.h>
16 #include <unistd.h> // for some IPC/network system calls
17 
18 #include <algorithm>
19 #include <cstddef>
20 
21 #include <boost/numeric/conversion/cast.hpp>
22 
23 #include <util/buffer.h>
24 #include <util/io_utilities.h>
25 
27 #include <asiolink/io_endpoint.h>
28 #include <asiolink/io_service.h>
29 #include <asiolink/tcp_endpoint.h>
30 
31 #include <exceptions/isc_assert.h>
32 
33 namespace isc {
34 namespace asiolink {
35 
39 class BufferTooLarge : public IOError {
40 public:
41  BufferTooLarge(const char* file, size_t line, const char* what) :
42  IOError(file, line, what) {}
43 };
44 
49 template <typename C>
50 class TCPSocket : public IOAsioSocket<C> {
51 private:
53  TCPSocket(const TCPSocket&);
54  TCPSocket& operator=(const TCPSocket&);
55 
56 public:
57 
63  TCPSocket(boost::asio::ip::tcp::socket& socket);
64 
71  TCPSocket(IOService& service);
72 
74  virtual ~TCPSocket();
75 
77  virtual int getNative() const {
78 #if BOOST_VERSION < 106600
79  return (socket_.native());
80 #else
81  return (socket_.native_handle());
82 #endif
83  }
84 
86  virtual int getProtocol() const {
87  return (IPPROTO_TCP);
88  }
89 
93  virtual bool isOpenSynchronous() const {
94  return (false);
95  }
96 
103  bool isUsable() const {
104  // If the socket is open it doesn't mean that it is still usable. The connection
105  // could have been closed on the other end. We have to check if we can still
106  // use this socket.
107  if (socket_.is_open()) {
108  // Remember the current non blocking setting.
109  const bool non_blocking_orig = socket_.non_blocking();
110  // Set the socket to non blocking mode. We're going to test if the socket
111  // returns would_block status on the attempt to read from it.
112  socket_.non_blocking(true);
113 
114  boost::system::error_code ec;
115  char data[2];
116 
117  // Use receive with message peek flag to avoid removing the data awaiting
118  // to be read.
119  socket_.receive(boost::asio::buffer(data, sizeof(data)),
120  boost::asio::socket_base::message_peek,
121  ec);
122 
123  // Revert the original non_blocking flag on the socket.
124  socket_.non_blocking(non_blocking_orig);
125 
126  // If the connection is alive we'd typically get would_block status code.
127  // If there are any data that haven't been read we may also get success
128  // status. We're guessing that try_again may also be returned by some
129  // implementations in some situations. Any other error code indicates a
130  // problem with the connection so we assume that the connection has been
131  // closed.
132  return (!ec || (ec.value() == boost::asio::error::try_again) ||
133  (ec.value() == boost::asio::error::would_block));
134  }
135 
136  return (false);
137  }
138 
146  virtual void open(const IOEndpoint* endpoint, C& callback);
147 
160  virtual void asyncSend(const void* data, size_t length,
161  const IOEndpoint* endpoint, C& callback);
162 
175  void asyncSend(const void* data, size_t length, C& callback);
176 
188  virtual void asyncReceive(void* data, size_t length, size_t offset,
189  IOEndpoint* endpoint, C& callback);
190 
206  virtual bool processReceivedData(const void* staging, size_t length,
207  size_t& cumulative, size_t& offset,
208  size_t& expected,
209  isc::util::OutputBufferPtr& outbuff);
210 
212  virtual void cancel();
213 
215  virtual void close();
216 
220  virtual boost::asio::ip::tcp::socket& getASIOSocket() const {
221  return (socket_);
222  }
223 
224 private:
228 
230  std::unique_ptr<boost::asio::ip::tcp::socket> socket_ptr_;
231 
233  boost::asio::ip::tcp::socket& socket_;
234 
248 
250  isc::util::OutputBufferPtr send_buffer_;
251 };
252 
253 // Constructor - caller manages socket
254 
255 template <typename C>
256 TCPSocket<C>::TCPSocket(boost::asio::ip::tcp::socket& socket) :
257  socket_ptr_(), socket_(socket), send_buffer_()
258 {
259 }
260 
261 // Constructor - create socket on the fly
262 
263 template <typename C>
265  socket_ptr_(new boost::asio::ip::tcp::socket(service.get_io_service())),
266  socket_(*socket_ptr_)
267 {
268 }
269 
270 // Destructor.
271 
272 template <typename C>
274 {
275 }
276 
277 // Open the socket.
278 
279 template <typename C> void
280 TCPSocket<C>::open(const IOEndpoint* endpoint, C& callback) {
281  // If socket is open on this end but has been closed by the peer,
282  // we need to reconnect.
283  if (socket_.is_open() && !isUsable()) {
284  close();
285  }
286  // Ignore opens on already-open socket. Don't throw a failure because
287  // of uncertainties as to what precedes when using asynchronous I/O.
288  // Also allows us a treat a passed-in socket as a self-managed socket.
289  if (!socket_.is_open()) {
290  if (endpoint->getFamily() == AF_INET) {
291  socket_.open(boost::asio::ip::tcp::v4());
292  } else {
293  socket_.open(boost::asio::ip::tcp::v6());
294  }
295 
296  // Set options on the socket:
297 
298  // Reuse address - allow the socket to bind to a port even if the port
299  // is in the TIMED_WAIT state.
300  socket_.set_option(boost::asio::socket_base::reuse_address(true));
301  }
302 
303  // Upconvert to a TCPEndpoint. We need to do this because although
304  // IOEndpoint is the base class of UDPEndpoint and TCPEndpoint, it does not
305  // contain a method for getting at the underlying endpoint type - that is in
307  isc_throw_assert(endpoint->getProtocol() == IPPROTO_TCP);
308  const TCPEndpoint* tcp_endpoint =
309  static_cast<const TCPEndpoint*>(endpoint);
310 
311  // Connect to the remote endpoint. On success, the handler will be
312  // called (with one argument - the length argument will default to
313  // zero).
314  socket_.async_connect(tcp_endpoint->getASIOEndpoint(), callback);
315 }
316 
317 // Send a message. Should never do this if the socket is not open, so throw
318 // an exception if this is the case.
319 
320 template <typename C> void
321 TCPSocket<C>::asyncSend(const void* data, size_t length, C& callback)
322 {
323  if (socket_.is_open()) {
324 
325  try {
326  send_buffer_.reset(new isc::util::OutputBuffer(length));
327  send_buffer_->writeData(data, length);
328 
329  // Send the data.
330  socket_.async_send(boost::asio::buffer(send_buffer_->getData(),
331  send_buffer_->getLength()),
332  callback);
333  } catch (const boost::numeric::bad_numeric_cast&) {
335  "attempt to send buffer larger than 64kB");
336  }
337 
338  } else {
340  "attempt to send on a TCP socket that is not open");
341  }
342 }
343 
344 template <typename C> void
345 TCPSocket<C>::asyncSend(const void* data, size_t length,
346  const IOEndpoint*, C& callback)
347 {
348  if (socket_.is_open()) {
349 
353  try {
355  uint16_t count = boost::numeric_cast<uint16_t>(length);
356 
358  send_buffer_.reset(new isc::util::OutputBuffer(length + 2));
359  send_buffer_->writeUint16(count);
360  send_buffer_->writeData(data, length);
361 
363  socket_.async_send(boost::asio::buffer(send_buffer_->getData(),
364  send_buffer_->getLength()), callback);
365  } catch (const boost::numeric::bad_numeric_cast&) {
367  "attempt to send buffer larger than 64kB");
368  }
369 
370  } else {
372  "attempt to send on a TCP socket that is not open");
373  }
374 }
375 
376 // Receive a message. Note that the "offset" argument is used as an index
377 // into the buffer in order to decide where to put the data. It is up to the
378 // caller to initialize the data to zero
379 template <typename C> void
380 TCPSocket<C>::asyncReceive(void* data, size_t length, size_t offset,
381  IOEndpoint* endpoint, C& callback)
382 {
383  if (socket_.is_open()) {
384  // Upconvert to a TCPEndpoint. We need to do this because although
385  // IOEndpoint is the base class of UDPEndpoint and TCPEndpoint, it
386  // does not contain a method for getting at the underlying endpoint
387  // type - that is in the derived class and the two classes differ on
388  // return type.
389  isc_throw_assert(endpoint->getProtocol() == IPPROTO_TCP);
390  TCPEndpoint* tcp_endpoint = static_cast<TCPEndpoint*>(endpoint);
391 
392  // Write the endpoint details from the communications link. Ideally
393  // we should make IOEndpoint assignable, but this runs in to all sorts
394  // of problems concerning the management of the underlying Boost
395  // endpoint (e.g. if it is not self-managed, is the copied one
396  // self-managed?) The most pragmatic solution is to let Boost take care
397  // of everything and copy details of the underlying endpoint.
398  tcp_endpoint->getASIOEndpoint() = socket_.remote_endpoint();
399 
400  // Ensure we can write into the buffer and if so, set the pointer to
401  // where the data will be written.
402  if (offset >= length) {
403  isc_throw(BufferOverflow, "attempt to read into area beyond end of "
404  "TCP receive buffer");
405  }
406  void* buffer_start = static_cast<void*>(static_cast<uint8_t*>(data) + offset);
407 
408  // ... and kick off the read.
409  socket_.async_receive(boost::asio::buffer(buffer_start, length - offset), callback);
410 
411  } else {
413  "attempt to receive from a TCP socket that is not open");
414  }
415 }
416 
417 // Is the receive complete?
418 
419 template <typename C> bool
420 TCPSocket<C>::processReceivedData(const void* staging, size_t length,
421  size_t& cumulative, size_t& offset,
422  size_t& expected,
424 {
425  // Point to the data in the staging buffer and note how much there is.
426  const uint8_t* data = static_cast<const uint8_t*>(staging);
427  size_t data_length = length;
428 
429  // Is the number is "expected" valid? It won't be unless we have received
430  // at least two bytes of data in total for this set of receives.
431  if (cumulative < 2) {
432 
433  // "expected" is not valid. Did this read give us enough data to
434  // work it out?
435  cumulative += length;
436  if (cumulative < 2) {
437 
438  // Nope, still not valid. This must have been the first packet and
439  // was only one byte long. Tell the fetch code to read the next
440  // packet into the staging buffer beyond the data that is already
441  // there so that the next time we are called we have a complete
442  // TCP count.
443  offset = cumulative;
444  return (false);
445  }
446 
447  // Have enough data to interpret the packet count, so do so now.
448  expected = isc::util::readUint16(data, cumulative);
449 
450  // We have two bytes less of data to process. Point to the start of the
451  // data and adjust the packet size. Note that at this point,
452  // "cumulative" is the true amount of data in the staging buffer, not
453  // "length".
454  data += 2;
455  data_length = cumulative - 2;
456  } else {
457 
458  // Update total amount of data received.
459  cumulative += length;
460  }
461 
462  // Regardless of anything else, the next read goes into the start of the
463  // staging buffer.
464  offset = 0;
465 
466  // Work out how much data we still have to put in the output buffer. (This
467  // could be zero if we have just interpreted the TCP count and that was
468  // set to zero.)
469  if (expected >= outbuff->getLength()) {
470 
471  // Still need data in the output packet. Copy what we can from the
472  // staging buffer to the output buffer.
473  size_t copy_amount = std::min(expected - outbuff->getLength(), data_length);
474  outbuff->writeData(data, copy_amount);
475  }
476 
477  // We can now say if we have all the data.
478  return (expected == outbuff->getLength());
479 }
480 
481 // Cancel I/O on the socket. No-op if the socket is not open.
482 
483 template <typename C> void
485  if (socket_.is_open()) {
486  socket_.cancel();
487  }
488 }
489 
490 // Close the socket down. Can only do this if the socket is open and we are
491 // managing it ourself.
492 
493 template <typename C> void
495  if (socket_.is_open() && socket_ptr_) {
496  socket_.close();
497  }
498 }
499 
500 } // namespace asiolink
501 } // namespace isc
502 
503 #endif // TCP_SOCKET_H
virtual const char * what() const
Returns a C-style character string of the cause of the exception.
The OutputBuffer class is a buffer abstraction for manipulating mutable data.
Definition: buffer.h:294
#define isc_throw(type, stream)
A shortcut macro to insert known values into exception arguments.
#define isc_throw_assert(expr)
Replacement for assert() that throws if the expression is false.
Definition: isc_assert.h:18
boost::shared_ptr< OutputBuffer > OutputBufferPtr
Definition: buffer.h:603
uint16_t readUint16(const void *buffer, size_t length)
Read Unsigned 16-Bit Integer from Buffer.
Definition: io_utilities.h:28
Defines the logger used by the top-level component of kea-lfc.