gvpe/src/vpn_tcp.C

/*
    vpn_tcp.C -- handle the tcp part of the protocol.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation; either version 2 of the License, or
    (at your option) any later version.
 
    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.
 
    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc. 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

#include "config.h"

#if ENABLE_TCP

// tcp processing is extremely ugly, since the vpe protocol is simply
// designed for unreliable datagram networks. tcp is implemented by
// multiplexing packets over tcp. errors are completely ignored, as we
// rely on the higher level protocol to time out and reconnect.

#include <cstring>

#include <sys/types.h>
#include <sys/socket.h>
#include <sys/poll.h>
#include <sys/wait.h>
#include <netinet/in.h>
#include <sys/uio.h>
#include <arpa/inet.h>
#include <errno.h>
#include <time.h>
#include <unistd.h>

#include <map>
#include <unistd.h>
#include <fcntl.h>
#include <sys/poll.h>

#include "vpn.h"

struct tcp_connection;

struct lt_sockinfo
{
  bool operator()(const sockinfo *a, const sockinfo *b) const
  {
    return *a < *b;
  }
};

struct tcp_si_map : public map<const sockinfo *, tcp_connection *, lt_sockinfo> {
  void cleaner_cb (time_watcher &w); time_watcher cleaner;

  tcp_si_map ()
    : cleaner(this, &tcp_si_map::cleaner_cb)
    {
      cleaner.start (0);
    }
} tcp_si;

struct tcp_connection : io_watcher {
  tstamp last_activity;
  const sockinfo si;
  vpn &v;
  bool active; // this connection has been actively established
  enum { ERROR, IDLE, CONNECTING, ESTABLISHED } state;

  vpn_packet *r_pkt;
  u32 r_len, r_ofs;

  vpn_packet *w_pkt;
  u32 w_len, w_ofs;

  void tcpv4_ev (io_watcher &w, short revents);

  bool send_packet (vpn_packet *pkt, int tos);
  bool write_packet ();

  void error (); // abort conenction && cleanup

  operator tcp_si_map::value_type()
    {
      return tcp_si_map::value_type (&si, this);
    }

  tcp_connection (int fd_, const sockinfo &si_, vpn &v_);
  ~tcp_connection ();
};

void tcp_si_map::cleaner_cb (time_watcher &w)
{
  w.at = NOW + 600;
  tstamp to = NOW - ::conf.keepalive - 30 - 60;

  for (iterator i = begin (); i != end(); )
    if (i->second->last_activity >= to)
      ++i;
    else
      {
        erase (i);
        i = begin ();
      }
}

void
vpn::tcpv4_ev (io_watcher &w, short revents)
{
  if (revents & (POLLIN | POLLERR))
    {
      struct sockaddr_in sa;
      socklen_t sa_len = sizeof (sa);
      int len;

      int fd = accept (w.fd, (sockaddr *)&sa, &sa_len);

      if (fd >= 0)
        {
          sockinfo si(sa, PROT_TCPv4);

          slog (L_DEBUG, _("%s: accepted tcp connection"), (const char *)si);//D

          fcntl (fd, F_SETFL, O_NONBLOCK);

          tcp_connection *i = new tcp_connection (fd, si, *this);
          tcp_si.insert (*i);
        }
    }
}

bool
vpn::send_tcpv4_packet (vpn_packet *pkt, const sockinfo &si, int tos)
{
  tcp_si_map::iterator info = tcp_si.find (&si);

  tcp_connection *i;

  if (info == tcp_si.end ())
    {
      i = new tcp_connection (-1, si, *this);
      tcp_si.insert (*i);
    }
  else
    i = info->second;

  return i->send_packet (pkt, tos);
}

void tcp_connection::error ()
{
  if (fd >= 0)
    {
      close (fd);
      fd = -1;
    }

  delete r_pkt; r_pkt = 0;
  delete w_pkt; w_pkt = 0;

  stop ();
  state = active ? IDLE : ERROR;
}

bool
tcp_connection::write_packet ()
{
  ssize_t len;

  if (w_ofs < 2)
    {
      u16 plen = htons (w_pkt->len);

      iovec vec[2];
      vec[0].iov_base = ((u8 *)&plen) + w_ofs;
      vec[0].iov_len = 2 - w_ofs;
      vec[1].iov_base = &((*w_pkt)[0]);
      vec[1].iov_len = w_len - 2;

      len = writev (fd, vec, 2);
    }
  else
    len = write (fd, &((*w_pkt)[w_ofs - 2]), w_len);

  if (len > 0)
    {
      w_ofs += len;
      w_len -= len;

      return w_len == 0;
    }
  else if (len < 0 && (errno == EAGAIN || errno == EINTR))
    return false;
  else
    {
      error ();
      return false;
    }
}

void
tcp_connection::tcpv4_ev (io_watcher &w, short revents)
{
  last_activity = NOW;

  if (revents & (POLLERR | POLLHUP))
    {
      error ();
      return;
    }

  if (revents & POLLOUT)
    {
      if (state == CONNECTING)
        {
          state = ESTABLISHED;
          set (POLLIN);
        }
      else if (state == ESTABLISHED)
        {
          if (w_pkt)
            {
              if (write_packet ())
                {
                  delete w_pkt; w_pkt = 0;

                  set (POLLIN);
                }
            }
          else
            set (POLLIN);
        }
      else
        set (POLLIN);
    }

  if (revents & POLLIN)
    {
      for (;;)
        {
          if (!r_pkt)
            {
              r_pkt = new vpn_packet;
              r_ofs = 0;
              r_len = 2; // header
            }

          ssize_t len = read (fd, &((*r_pkt)[r_ofs < 2 ? r_ofs : r_ofs - 2]), r_len);

          if (len > 0)
            {
              r_len -= len;
              r_ofs += len;

              if (r_len == 0)
                {
                  if (r_ofs == 2)
                    {
                      r_len = ntohs (*(u16 *)&((*r_pkt)[0]));
                      r_pkt->len = r_len;

                      if (r_len > 0 && r_len < MAXSIZE)
                        continue;
                    }
                  else
                    {
                      v.recv_vpn_packet (r_pkt, si);
                      delete r_pkt;
                      r_pkt = 0;

                      continue;
                    }
                }
              else
                break;
            }
          else if (len < 0 && (errno == EINTR || errno == EAGAIN))
            break;

          error ();
          break;
        }
    }
}

bool
tcp_connection::send_packet (vpn_packet *pkt, int tos)
{
  last_activity = NOW;

  if (state == IDLE)
    {
      // woaw, the first lost packet ;)
      fd = socket (PF_INET, SOCK_STREAM, IPPROTO_TCP);

      if (fd >= 0)
        {
          fcntl (fd, F_SETFL, O_NONBLOCK);
          
          if (connect (fd, si.sav4 (), si.salenv4 ()) >= 0
              || errno == EINPROGRESS)
            {
              state = CONNECTING;
              start (fd, POLLOUT);
            }
          else
            close (fd);
        }
    }
  else if (state == ESTABLISHED)
    {
      // drop packet if the tcp write buffer is full. this *is* the
      // right thing to do, not using tcp *is* the right thing to do.
      if (!w_pkt)
        {
          // how this maps to the underlying tcp packets we don't know
          // and we don't care. at least we tried ;)
          setsockopt (fd, SOL_IP, IP_TOS, &tos, sizeof tos);

          w_pkt = pkt;
          w_ofs = 0;
          w_len = pkt->len + 2; // length + size header

          if (write_packet ())
            w_pkt = 0;
          else
            {
              w_pkt = new vpn_packet;
              w_pkt->set (*pkt);

              set (POLLIN | POLLOUT);
            }
        }
    }

  return state != ERROR;
}

tcp_connection::tcp_connection (int fd_, const sockinfo &si_, vpn &v_)
: v(v_), si(si_), io_watcher(this, &tcp_connection::tcpv4_ev)
{
  last_activity = NOW;
  r_pkt = 0;
  w_pkt = 0;
  fd = fd_;

  if (fd < 0)
    {
      active = true;
      state = IDLE;
    }
  else
    {
      active = false;
      state = ESTABLISHED;
      start (fd, POLLIN);
    }
}

tcp_connection::~tcp_connection ()
{
  error ();
}

#endif

Revision:	1.3
Committed:	Sun Apr 6 20:01:53 2003 UTC (21 years, 1 month ago) by pcg
Content type:	text/plain
Branch:	MAIN
Changes since 1.2:	+96 -27 lines
Log Message:	* empty log message *
#	Content
1	/*
2	vpn_tcp.C -- handle the tcp part of the protocol.
3
4	This program is free software; you can redistribute it and/or modify
5	it under the terms of the GNU General Public License as published by
6	the Free Software Foundation; either version 2 of the License, or
7	(at your option) any later version.
8
9	This program is distributed in the hope that it will be useful,
10	but WITHOUT ANY WARRANTY; without even the implied warranty of
11	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12	GNU General Public License for more details.
13
14	You should have received a copy of the GNU General Public License
15	along with this program; if not, write to the Free Software
16	Foundation, Inc. 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17	*/
18
19	#include "config.h"
20
21	#if ENABLE_TCP
22
23	// tcp processing is extremely ugly, since the vpe protocol is simply
24	// designed for unreliable datagram networks. tcp is implemented by
25	// multiplexing packets over tcp. errors are completely ignored, as we
26	// rely on the higher level protocol to time out and reconnect.
27
28	#include <cstring>
29
30	#include <sys/types.h>
31	#include <sys/socket.h>
32	#include <sys/poll.h>
33	#include <sys/wait.h>
34	#include <netinet/in.h>
35	#include <sys/uio.h>
36	#include <arpa/inet.h>
37	#include <errno.h>
38	#include <time.h>
39	#include <unistd.h>
40
41	#include <map>
42	#include <unistd.h>
43	#include <fcntl.h>
44	#include <sys/poll.h>
45
46	#include "vpn.h"
47
48	struct tcp_connection;
49
50	struct lt_sockinfo
51	{
52	bool operator()(const sockinfo a, const sockinfo b) const
53	{
54	return a < b;
55	}
56	};
57
58	struct tcp_si_map : public map<const sockinfo , tcp_connection , lt_sockinfo> {
59	void cleaner_cb (time_watcher &w); time_watcher cleaner;
60
61	tcp_si_map ()
62	: cleaner(this, &tcp_si_map::cleaner_cb)
63	{
64	cleaner.start (0);
65	}
66	} tcp_si;
67
68	struct tcp_connection : io_watcher {
69	tstamp last_activity;
70	const sockinfo si;
71	vpn &v;
72	bool active; // this connection has been actively established
73	enum { ERROR, IDLE, CONNECTING, ESTABLISHED } state;
74
75	vpn_packet *r_pkt;
76	u32 r_len, r_ofs;
77
78	vpn_packet *w_pkt;
79	u32 w_len, w_ofs;
80
81	void tcpv4_ev (io_watcher &w, short revents);
82
83	bool send_packet (vpn_packet *pkt, int tos);
84	bool write_packet ();
85
86	void error (); // abort conenction && cleanup
87
88	operator tcp_si_map::value_type()
89	{
90	return tcp_si_map::value_type (&si, this);
91	}
92
93	tcp_connection (int fd_, const sockinfo &si_, vpn &v_);
94	~tcp_connection ();
95	};
96
97	void tcp_si_map::cleaner_cb (time_watcher &w)
98	{
99	w.at = NOW + 600;
100	tstamp to = NOW - ::conf.keepalive - 30 - 60;
101
102	for (iterator i = begin (); i != end(); )
103	if (i->second->last_activity >= to)
104	++i;
105	else
106	{
107	erase (i);
108	i = begin ();
109	}
110	}
111
112	void
113	vpn::tcpv4_ev (io_watcher &w, short revents)
114	{
115	if (revents & (POLLIN \| POLLERR))
116	{
117	struct sockaddr_in sa;
118	socklen_t sa_len = sizeof (sa);
119	int len;
120
121	int fd = accept (w.fd, (sockaddr *)&sa, &sa_len);
122
123	if (fd >= 0)
124	{
125	sockinfo si(sa, PROT_TCPv4);
126
127	slog (L_DEBUG, _("%s: accepted tcp connection"), (const char *)si);//D
128
129	fcntl (fd, F_SETFL, O_NONBLOCK);
130
131	tcp_connection i = new tcp_connection (fd, si, this);
132	tcp_si.insert (*i);
133	}
134	}
135	}
136
137	bool
138	vpn::send_tcpv4_packet (vpn_packet *pkt, const sockinfo &si, int tos)
139	{
140	tcp_si_map::iterator info = tcp_si.find (&si);
141
142	tcp_connection *i;
143
144	if (info == tcp_si.end ())
145	{
146	i = new tcp_connection (-1, si, *this);
147	tcp_si.insert (*i);
148	}
149	else
150	i = info->second;
151
152	return i->send_packet (pkt, tos);
153	}
154
155	void tcp_connection::error ()
156	{
157	if (fd >= 0)
158	{
159	close (fd);
160	fd = -1;
161	}
162
163	delete r_pkt; r_pkt = 0;
164	delete w_pkt; w_pkt = 0;
165
166	stop ();
167	state = active ? IDLE : ERROR;
168	}
169
170	bool
171	tcp_connection::write_packet ()
172	{
173	ssize_t len;
174
175	if (w_ofs < 2)
176	{
177	u16 plen = htons (w_pkt->len);
178
179	iovec vec[2];
180	vec[0].iov_base = ((u8 *)&plen) + w_ofs;
181	vec[0].iov_len = 2 - w_ofs;
182	vec[1].iov_base = &((*w_pkt)[0]);
183	vec[1].iov_len = w_len - 2;
184
185	len = writev (fd, vec, 2);
186	}
187	else
188	len = write (fd, &((*w_pkt)[w_ofs - 2]), w_len);
189
190	if (len > 0)
191	{
192	w_ofs += len;
193	w_len -= len;
194
195	return w_len == 0;
196	}
197	else if (len < 0 && (errno == EAGAIN \|\| errno == EINTR))
198	return false;
199	else
200	{
201	error ();
202	return false;
203	}
204	}
205
206	void
207	tcp_connection::tcpv4_ev (io_watcher &w, short revents)
208	{
209	last_activity = NOW;
210
211	if (revents & (POLLERR \| POLLHUP))
212	{
213	error ();
214	return;
215	}
216
217	if (revents & POLLOUT)
218	{
219	if (state == CONNECTING)
220	{
221	state = ESTABLISHED;
222	set (POLLIN);
223	}
224	else if (state == ESTABLISHED)
225	{
226	if (w_pkt)
227	{
228	if (write_packet ())
229	{
230	delete w_pkt; w_pkt = 0;
231
232	set (POLLIN);
233	}
234	}
235	else
236	set (POLLIN);
237	}
238	else
239	set (POLLIN);
240	}
241
242	if (revents & POLLIN)
243	{
244	for (;;)
245	{
246	if (!r_pkt)
247	{
248	r_pkt = new vpn_packet;
249	r_ofs = 0;
250	r_len = 2; // header
251	}
252
253	ssize_t len = read (fd, &((*r_pkt)[r_ofs < 2 ? r_ofs : r_ofs - 2]), r_len);
254
255	if (len > 0)
256	{
257	r_len -= len;
258	r_ofs += len;
259
260	if (r_len == 0)
261	{
262	if (r_ofs == 2)
263	{
264	r_len = ntohs ((u16 )&((*r_pkt)[0]));
265	r_pkt->len = r_len;
266
267	if (r_len > 0 && r_len < MAXSIZE)
268	continue;
269	}
270	else
271	{
272	v.recv_vpn_packet (r_pkt, si);
273	delete r_pkt;
274	r_pkt = 0;
275
276	continue;
277	}
278	}
279	else
280	break;
281	}
282	else if (len < 0 && (errno == EINTR \|\| errno == EAGAIN))
283	break;
284
285	error ();
286	break;
287	}
288	}
289	}
290
291	bool
292	tcp_connection::send_packet (vpn_packet *pkt, int tos)
293	{
294	last_activity = NOW;
295
296	if (state == IDLE)
297	{
298	// woaw, the first lost packet ;)
299	fd = socket (PF_INET, SOCK_STREAM, IPPROTO_TCP);
300
301	if (fd >= 0)
302	{
303	fcntl (fd, F_SETFL, O_NONBLOCK);
304
305	if (connect (fd, si.sav4 (), si.salenv4 ()) >= 0
306	\|\| errno == EINPROGRESS)
307	{
308	state = CONNECTING;
309	start (fd, POLLOUT);
310	}
311	else
312	close (fd);
313	}
314	}
315	else if (state == ESTABLISHED)
316	{
317	// drop packet if the tcp write buffer is full. this is the
318	// right thing to do, not using tcp is the right thing to do.
319	if (!w_pkt)
320	{
321	// how this maps to the underlying tcp packets we don't know
322	// and we don't care. at least we tried ;)
323	setsockopt (fd, SOL_IP, IP_TOS, &tos, sizeof tos);
324
325	w_pkt = pkt;
326	w_ofs = 0;
327	w_len = pkt->len + 2; // length + size header
328
329	if (write_packet ())
330	w_pkt = 0;
331	else
332	{
333	w_pkt = new vpn_packet;
334	w_pkt->set (*pkt);
335
336	set (POLLIN \| POLLOUT);
337	}
338	}
339	}
340
341	return state != ERROR;
342	}
343
344	tcp_connection::tcp_connection (int fd_, const sockinfo &si_, vpn &v_)
345	: v(v_), si(si_), io_watcher(this, &tcp_connection::tcpv4_ev)
346	{
347	last_activity = NOW;
348	r_pkt = 0;
349	w_pkt = 0;
350	fd = fd_;
351
352	if (fd < 0)
353	{
354	active = true;
355	state = IDLE;
356	}
357	else
358	{
359	active = false;
360	state = ESTABLISHED;
361	start (fd, POLLIN);
362	}
363	}
364
365	tcp_connection::~tcp_connection ()
366	{
367	error ();
368	}
369
370	#endif
371