net.c

MySQL的ODBC接口程序源代码

/* Copyright Abandoned 1996 TCX DataKonsult AB & Monty Program KB & Detron HB
   This file is public domain and comes with NO WARRANTY of any kind */

/* Write and read of logical packets to/from socket
** Writes are cached into net_buffer_length big packets.
** Read packets are reallocated dynamicly when reading big packets.
** Each logical packet has the following pre-info:
** 3 byte length & 1 byte package-number.
*/

#ifdef _WIN32
#include <winsock.h>
#endif
#include <global.h>
#include <my_sys.h>
#include <m_string.h>
#include "mysql.h"
#include <signal.h>
#include <errno.h>
#include <sys/types.h>
#if !defined(__WIN32__) && !defined(MSDOS)
#include <sys/socket.h>
#endif
#if !defined(MSDOS) && !defined(__WIN32__) && !defined(HAVE_BROKEN_NETINET_INCLUDES)
#include <netinet/in_systm.h>
#include <netinet/in.h>
#include <netinet/ip.h>
#include <netinet/tcp.h>
#endif
#ifdef MYSQL_SERVER
#include "my_pthread.h"
#include "thr_alarm.h"
#endif

#if defined(MSDOS) || defined(__WIN32__)
#define raw_net_read(A,B,C) recv((A),(B),(C),0)
#define raw_net_write(A,B,C) send((A),(B),(C),0)
#ifdef __WIN32__
#undef errno
#undef EINTR
#undef EAGAIN
#define errno WSAGetLastError()
#define EINTR  WSAEINTR
#define EAGAIN WSAEINPROGRESS
#endif
#else /* unix */
#define raw_net_read(A,B,C) read((A),(B),(C))
#define raw_net_write(A,B,C) write((A),(B),(C))
#endif
#ifndef EWOULDBLOCK
#define EWOULDBLOCK EAGAIN
#endif

/*
** Give error if a too big packet is found
** The server can change this with the -O switch, but because the client
** can't normally do this the client should have a bigger max-buffer.
*/

#ifdef MYSQL_SERVER
ulong max_allowed_packet=65536;
#else
ulong max_allowed_packet=1024*1024L;
#endif
ulong net_buffer_length=8192;	/* Default length. Enlarged if necessary */

static int net_write_buff(NET *net,const byte *packet,uint len);


	/* Init with packet info */

int my_net_init(NET *net,Socket fd)
{
  if (!(net->buff=(uchar*) my_malloc(net_buffer_length,MYF(MY_WME))))
    return 1;

  if (net_buffer_length > max_allowed_packet)
    max_allowed_packet=net_buffer_length;
  net->buff_end=net->buff+(net->max_packet=net_buffer_length);
  net->fd=fd;
  net->error=net->return_errno=0;
  net->timeout=30;				/* Timeout for read */
  net->pkt_nr=0;
  net->write_pos=net->buff;
  net->last_error[0]=0;
#if defined(MYSQL_SERVER) && !defined(__WIN32__)
  if (fd)
  {
    extern uint test_flags;			/* QQ */
    net->fcntl=(fcntl(net->fd, F_GETFL) |
		(!(test_flags & 8) ? O_NONBLOCK : 0));
    (void) fcntl(net->fd, F_SETFL, net->fcntl);
  }
#endif
#ifdef IPTOS_THROUGHPUT				/* For FreeBSD */
  {
    int tos = IPTOS_THROUGHPUT;
    if (!setsockopt(fd, IPPROTO_IP, IP_TOS, (void*) &tos, sizeof(tos)))
    {
      int nodelay = 1;
      if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (void*) &nodelay,
		     sizeof(nodelay)))
      {
	DBUG_PRINT("warning",("Couldn't set socket option for fast send"));
      }
    }
  }
#endif
  return 0;
}

void net_end(NET *net)
{
  if (net->buff)
  {
    my_free((gptr) net->buff,MYF(0));
    net->buff=0;
  }
}

	/* Remove unwanted characters from connection */

#ifndef MYSQL_SERVER
void net_clear(NET *net)
{
#if !defined(MSDOS) && !defined(__WIN32__)
  int old_fcntl=fcntl(net->fd, F_GETFL);
  int count;

  if (!(old_fcntl & O_NONBLOCK))
    (void) fcntl(net->fd, F_SETFL, old_fcntl | O_NONBLOCK);
  while ((count=(int) raw_net_read(net->fd,net->buff,net->max_packet)) > 0)
  {
    DBUG_PRINT("info",("skipped %d bytes from file: %d",count,net->fd));
  }
  if (!(old_fcntl & O_NONBLOCK))
    (void) fcntl(net->fd, F_SETFL, old_fcntl);
#else
  ulong arg;
  arg=1; ioctlsocket(net->fd,FIONBIO,&arg);
  while ((int) raw_net_read(net->fd,net->buff,net->max_packet) > 0) ;
  arg=0; ioctlsocket(net->fd,FIONBIO,&arg);
#endif
  net->pkt_nr=0;				/* Ready for new command */
  net->write_pos=net->buff;
}
#endif

	/* Flush write_buffer if not empty. */

int net_flush(NET *net)
{
  int error=0;
  if (net->buff != net->write_pos)
  {
    error=net_real_write(net,(byte*) net->buff,
			 (uint) (net->write_pos - net->buff));
    net->write_pos=net->buff;
  }
  return error;
}


/*****************************************************************************
** Write something to server/clinet buffer
*****************************************************************************/


/*
** Write a logical packet with packet header
** Format: Packet length (3 bytes), packet number(1 byte)
*/

int
my_net_write(NET *net,const byte *packet,uint len)
{
  uchar buff[4];
  int3store(buff,len);
  buff[3]=(uchar) (net->pkt_nr++);
  if (net_write_buff(net,(char*) buff,4))
    return 1;
  return net_write_buff(net,packet,len);
}

int
net_write_command(NET *net,uchar command,const byte *packet,uint len)
{
  uchar buff[5];
  uint length=len+1;				/* One extra byte for command */

  int3store(buff,length);
  buff[3]=(uchar) (net->pkt_nr++);
  buff[4]=command;
  if (!net->buff)
  {						/* If net not initiated */
    return test(net_real_write(net,(byte*) buff,5) ||
		net_real_write(net,packet,len));
  }
  if (net_write_buff(net,(char*) buff,5))
    return 1;
  return test(net_write_buff(net,packet,len) || net_flush(net));
}


static int
net_write_buff(NET *net,const byte *packet,uint len)
{
  uint left_length=(uint) (net->buff_end - net->write_pos);

  while (len > left_length)
  {
    memcpy((byte*) net->write_pos,packet,left_length);
    if (net_real_write(net,(byte*) net->buff,net->max_packet))
      return 1;
    net->write_pos=net->buff;
    packet+=left_length;
    len-=left_length;
    left_length=net->max_packet;
  }
  memcpy((byte*) net->write_pos,packet,len);
  net->write_pos+=len;
  return 0;
}


#ifdef MYSQL_SERVER

/*  Read and write using timeouts */

int
net_real_write(NET *net,const byte *packet,uint len)
{
  int length;
  byte *pos,*end;
  thr_alarm_t alarmed;
  uint retry_count=0;

#ifdef __WIN32__
  alarmed=thr_alarm(60);			/* blocking read */
#else
  thr_alarm_init(&alarmed);
#endif
  pos=(byte*) packet; end=pos+len;
  while (pos != end)
  {
    if ((int) (length=raw_net_write(net->fd,pos,(size_t) (end-pos))) <= 0)
    {
#ifndef __WIN32__
      if ((errno == EAGAIN || errno == EWOULDBLOCK || length == 0)
	  && !thr_alarm_in_use(alarmed))
      {
	if (!thr_alarm(&alarmed,60))		/* Don't wait too long */
	{
	  while (fcntl(net->fd, F_SETFL, net->fcntl & ~ O_NONBLOCK) < 0)
	  {
	    if (errno == EINTR && retry_count++ < 10)
	      continue;
	    fprintf(stderr,
		    "%s: my_net_write: fcntl returned error %d, aborting thread\n",
		    my_progname,errno);
	    net->error=1;			/* Close socket */
	    goto end;
	  }
	  retry_count=0;
	  continue;
	}
      }
      else
#endif
	if (thr_alarm_in_use(alarmed) && !thr_got_alarm(alarmed) &&
	    (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK))
      {
	if (retry_count++ < 10)
	    continue;
	  fprintf(stderr, "%s: net_real_write looped, aborting thread\n",
		  my_progname);
      }
      net->error=1;				/* Close socket */
      break;
    }
    pos+=length;
  }
 end:
  if (thr_alarm_in_use(alarmed))
  {
    thr_end_alarm(&alarmed);
#ifdef HAVE_FCNTL
    (void) fcntl(net->fd, F_SETFL, net->fcntl);
#endif
  }
  return (int) (pos != end);
}

#else

int
net_real_write(NET *net,const byte *packet,uint len)
{
  int length;
  byte *pos,*end;
  bool alarmed=0;

  pos=(byte*) packet; end=pos+len;
  while (pos != end)
  {
    if ((int) (length=raw_net_write(net->fd,pos,(size_t) (end-pos))) <= 0)
    {
      if ((errno == EAGAIN || errno == EINTR || errno == EWOULDBLOCK ||
	   length == 0) && !alarmed)
      {
	alarmed=1;
#ifdef HAVE_FCNTL
	(void) fcntl(net->fd, F_SETFL, net->fcntl & ~ O_NONBLOCK);
#endif
	continue;
      }
#ifdef THREAD_SAFE_CLIENT
      if (errno == EINTR)
      {
	DBUG_PRINT("warning",("Interrupted write. Retrying..."));
	continue;
      }
#endif
      break;
    }
    pos+=length;
  }
#ifdef HAVE_FCNTL
  if (alarmed)
    (void) fcntl(net->fd, F_SETFL, net->fcntl);
#endif
  return (int) (pos != end);
}
#endif


/*****************************************************************************
** Read something from server/clinet
*****************************************************************************/

#ifdef MYSQL_SERVER

uint
my_net_read(NET *net)
{
  uchar *pos;
  ulong len,remain;
  long length;
  uint i,retry_count=0;
  thr_alarm_t alarmed;

  len=packet_error;
  remain = 4;
  pos=net->buff;

#ifdef __WIN32__
  alarmed=thr_alarm(net->timeout);		/* blocking read */
#else
  thr_alarm_init(&alarmed);
#endif
  for (i=0 ; i < 2 ; i++)
  {
    while (remain > 0)
    {
      errno=0;					/* For linux */
      if ((int) (length=raw_net_read(net->fd,(char*) pos,remain)) <= 0)
      {
#ifndef __WIN32__
	if ((errno == EAGAIN || errno == EWOULDBLOCK || length == 0)
	    && !thr_alarm_in_use(alarmed))
	{
	  if (!thr_alarm(&alarmed,net->timeout)) /* Don't wait too long */
	  {
	    while (fcntl(net->fd, F_SETFL, net->fcntl & ~ O_NONBLOCK) < 0)
	    {
	      if (errno == EINTR && retry_count++ < 10)
		continue;
	      DBUG_PRINT("error",("fcntl returned error %d, aborting thread",
				  errno));
	      fprintf(stderr,
		      "%s: my_net_read: fcntl returned error %d, aborting thread\n",
		      my_progname,errno);
	      len= packet_error;
	      net->error=1;			/* Close socket */
	      goto end;
	    }
	    retry_count=0;
	    continue;
	  }
	}
#endif
	if (thr_alarm_in_use(alarmed) && !thr_got_alarm(alarmed) &&
	    (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK))
	{					/* Probably in MIT threads */
	  if (retry_count++ < 10)
	    continue;
	  fprintf(stderr, "%s: my_net_read: looped with error %d, aborting thread\n",
		  my_progname,errno);
	}
	DBUG_PRINT("error",("Got error %d reading socket",errno));
	len= packet_error;
	net->error=1;				/* Close socket */
	goto end;
      }
      remain -= (ulong) length;
      pos+= (ulong) length;
    }
    if (i == 0)
    {					/* First parts is packet length */
      if ((len=uint3korr(net->buff)) >= max_allowed_packet)
      {
	DBUG_PRINT("error",("Packet too large (%ld)", len));
	fprintf(stderr,"Packet too large (%ld)\n", len);
	len= packet_error;		/* Return error */
	goto end;
      }
      if (net->buff[3] != (uchar) net->pkt_nr)
      {
	DBUG_PRINT("error",("Packets out of order (Found: %d, expected %d)",
			    (int) net->buff[3],net->pkt_nr));
	fprintf(stderr,"Packets out of order (Found: %d, expected %d)\n",
		(int) net->buff[3],net->pkt_nr);
	len= packet_error;
	goto end;
      }
      net->pkt_nr++;
      if (len >= net->max_packet)
      {					/* Alloc bigger package */
	uint pkt_length=(len+IO_SIZE) & ~(IO_SIZE-1);
	uchar *buff=(uchar*) my_realloc((char*) net->buff,pkt_length,
					MYF(MY_WME));
	if (!buff)
	{
	  len=packet_error;
	  goto end;
	}
	net->buff=net->write_pos=buff;
	net->buff_end=buff+(net->max_packet=pkt_length);
      }
      pos=net->buff;
      remain = len;
    }
  }
  *pos = 0;				/* Safeguard */

end:
  if (thr_alarm_in_use(alarmed))
  {
    thr_end_alarm(&alarmed);
#ifdef HAVE_FCNTL
    (void) fcntl(net->fd, F_SETFL, net->fcntl);
#endif
  }
  return(len);
}

#else /* !MYSQL_SERVER */

uint my_net_read(NET *net)
{
  uchar *pos,wrong_packet_nr;
  ulong len,remain;
  long length;
  uint i;
  bool alarmed=0,error_packet=0;

  len= packet_error;
  remain = 4;
  pos=net->buff;			/* net->packet -4 */
  for (i=0 ; i < 2 ; i++)
  {
    while (remain > 0)
    {
      if ((int) (length=raw_net_read(net->fd,pos,remain)) <= 0)
      {
	if ((errno == EAGAIN || errno == EINTR || errno == EWOULDBLOCK ||
	     length == 0) && !alarmed)
	{
	  alarmed=1;
#ifdef HAVE_FCNTL
	  (void) fcntl(net->fd, F_SETFL, net->fcntl & ~ O_NONBLOCK);
#endif
	  continue;
	}
#ifdef THREAD_SAFE_CLIENT
	if (errno == EINTR)
	{
	  DBUG_PRINT("warning",("Interrupted read. Retrying..."));
	  continue;
	}
#endif
	DBUG_PRINT("error",("Couldn't read packet: remain: %d  errno: %d  length: %d  alarmed: %d", remain,errno,length,alarmed));
	len= packet_error;
	goto end;
      }
      remain -= (ulong) length;
      pos+= (ulong) length;
    }
    if (i == 0)
    {					/* First parts is packet length */
      if ((len=uint3korr(net->buff)) >= max_allowed_packet)
      {
	DBUG_PRINT("error",("Packet too large (%ld)",len));
	fprintf(stderr,"Packet too large (%ld)\n", len);
	return packet_error;		/* Return error */
      }
      if (net->buff[3] != (uchar) net->pkt_nr)
      {
	error_packet=1;			/* Probably error out of sync */
	wrong_packet_nr=net->pkt_nr;
	net->pkt_nr=net->buff[3];
      }
      else
	net->pkt_nr++;
      if (len >= net->max_packet)
      {					/* Alloc bigger package */
	uint pkt_length=(len+IO_SIZE) & ~(IO_SIZE-1);
	uchar *buff=(uchar*) my_realloc((char*) net->buff,pkt_length,
					MYF(MY_WME));
	if (!buff)
	{
	  len=packet_error;
	  break;
	}
	net->buff=net->write_pos=buff;
	net->buff_end=buff+(net->max_packet=pkt_length);
      }
      pos=net->buff;
      remain = len;
    }
  }
  if (error_packet && net->buff[0] != (uchar) 255)
  {
    DBUG_PRINT("error",("Packets out of order (Found: %d, expected %d)\n",
			(int) (uchar) net->pkt_nr,(int) wrong_packet_nr));
    fprintf(stderr,"Packets out of order (Found: %d, expected %d)\n",
	    (int) (uchar) net->pkt_nr,(int) wrong_packet_nr);
    len=packet_error;
  }
end:
#ifdef HAVE_FCNTL
  if (alarmed)
    (void) fcntl(net->fd, F_SETFL, net->fcntl);
#endif
  *pos = 0;				/* Safeguard */
  return(len);
}

#endif