Coro-Mysql/Mysql.xs

#include <sys/errno.h>
#include <unistd.h>
#include <fcntl.h>

#include <mysql.h>

#include "EXTERN.h"
#include "perl.h"
#include "XSUB.h"

typedef U16 uint16;

/* cached function gv's */
static CV *readable, *writable;

#include "violite.h"

#define DESC_OFFSET 22

#define CoMy_MAGIC 0x436f4d79

typedef struct {
  int magic;
  SV *corosocket;
  int bufofs, bufcnt;
  char buf[VIO_READ_BUFFER_SIZE];
} ourdata;

#define OURDATAPTR (*((ourdata **)((vio)->desc + DESC_OFFSET)))

static int
our_read (Vio *vio, gptr p, int len)
{
  ourdata *our = OURDATAPTR;

  if (!our->bufcnt)
    {
      int rd;
      my_bool dummy;

      vio->vioblocking (vio, 0, &dummy);

      for (;;)
        {
          rd = recv (vio->sd, our->buf, sizeof (our->buf), 0);

          if (rd >= 0 || errno != EAGAIN)
            break;

          {
            dSP;
            PUSHMARK (SP);
            XPUSHs (our->corosocket);
            PUTBACK;
            call_sv ((SV *)readable, G_VOID | G_DISCARD);
          }
        }

      if (rd <= 0)
        return rd;

      our->bufcnt = rd;
      our->bufofs = 0;
    }

  if (our->bufcnt < len)
    len = our->bufcnt;

  memcpy (p, our->buf + our->bufofs, len);
  our->bufofs += len;
  our->bufcnt -= len;

  return len;
}

static int
our_write (Vio *vio, const gptr p, int len)
{
  char *ptr = (char *)p;
  my_bool dummy;

  vio->vioblocking (vio, 0, &dummy);

  while (len > 0)
    {
      int wr = send (vio->sd, ptr, len, 0);

      if (wr > 0)
        {
          ptr += wr;
          len -= wr;
        }
      else if (errno == EAGAIN)
        {
          dSP;
          PUSHMARK (SP);
          XPUSHs (OURDATAPTR->corosocket);
          PUTBACK;
          call_sv ((SV *)writable, G_VOID | G_DISCARD);
        }
      else if (ptr == (char *)p)
        return -1;
      else
        break;
    }

  return ptr - (char *)p;
}

MODULE = Coro::Mysql            PACKAGE = Coro::Mysql

BOOT:
{
  readable = get_cv ("Coro::Mysql::readable", 0);
  writable = get_cv ("Coro::Mysql::writable", 0);
}

PROTOTYPES: ENABLE

void
_patch (IV sock, int fd, SV *corosocket)
        CODE:
{
        MYSQL *my = (MYSQL *)sock;
        Vio *vio = my->net.vio;
        ourdata *our;

        if (fd != my->net.fd)
          croak ("DBD::mysql fd and libmysql disagree - library mismatch, unsupported transport or API changes?");

        if (fd != vio->sd)
          croak ("DBD::mysql fd and vio-sd disagree - library mismatch, unsupported transport or API changes?");

        if (vio->write != vio_write)
          croak ("vio.write has unexpected content - library mismatch, unsupported transport or API changes?");

        if (vio->read != vio_read && vio->read != vio_read_buff)
          croak ("vio.read has unexpected content - library mismatch, unsupported transport or API changes?");

        Newz (0, our, 1, ourdata);
        our->magic = CoMy_MAGIC;
        our->corosocket = newSVsv (corosocket);

        vio->desc [DESC_OFFSET - 1] = 0;
        OURDATAPTR = our;

        vio->write = our_write;
        vio->read  = our_read;
}

void
_unpatch (IV sock)
        CODE:
{
        MYSQL *my = (MYSQL *)sock;
        Vio *vio = my->net.vio;
        my_bool dummy;

        if (vio->read != our_read)
          croak ("vio.read has unexpected content during unpatch - wtf?");

        if (vio->write != our_write)
          croak ("vio.write has unexpected content during unpatch - wtf?");

        SvREFCNT_dec (OURDATAPTR->corosocket);

        Safefree (OURDATAPTR);

        vio->read  = vio_read;
        vio->write = vio_write;
}


Revision:	1.2
Committed:	Tue Jun 16 17:28:00 2009 UTC (14 years, 11 months ago) by root
Branch:	MAIN
Changes since 1.1:	+3 -0 lines
Log Message:	* empty log message *
#	User	Rev	Content
1	root	1.1	#include <sys/errno.h>
2			#include <unistd.h>
3			#include <fcntl.h>
4
5			#include <mysql.h>
6
7			#include "EXTERN.h"
8			#include "perl.h"
9			#include "XSUB.h"
10
11			typedef U16 uint16;
12
13			/* cached function gv's */
14			static CV readable, writable;
15
16			#include "violite.h"
17
18			#define DESC_OFFSET 22
19
20			#define CoMy_MAGIC 0x436f4d79
21
22			typedef struct {
23			int magic;
24			SV *corosocket;
25			int bufofs, bufcnt;
26			char buf[VIO_READ_BUFFER_SIZE];
27			} ourdata;
28
29			#define OURDATAPTR (((ourdata *)((vio)->desc + DESC_OFFSET)))
30
31			static int
32			our_read (Vio *vio, gptr p, int len)
33			{
34			ourdata *our = OURDATAPTR;
35
36			if (!our->bufcnt)
37			{
38			int rd;
39			my_bool dummy;
40
41			vio->vioblocking (vio, 0, &dummy);
42
43			for (;;)
44			{
45			rd = recv (vio->sd, our->buf, sizeof (our->buf), 0);
46
47			if (rd >= 0 \|\| errno != EAGAIN)
48			break;
49
50			{
51			dSP;
52			PUSHMARK (SP);
53			XPUSHs (our->corosocket);
54			PUTBACK;
55			call_sv ((SV *)readable, G_VOID \| G_DISCARD);
56			}
57			}
58
59			if (rd <= 0)
60			return rd;
61
62			our->bufcnt = rd;
63			our->bufofs = 0;
64			}
65
66			if (our->bufcnt < len)
67			len = our->bufcnt;
68
69			memcpy (p, our->buf + our->bufofs, len);
70			our->bufofs += len;
71			our->bufcnt -= len;
72
73			return len;
74			}
75
76			static int
77			our_write (Vio *vio, const gptr p, int len)
78			{
79			char ptr = (char )p;
80			my_bool dummy;
81
82			vio->vioblocking (vio, 0, &dummy);
83
84			while (len > 0)
85			{
86			int wr = send (vio->sd, ptr, len, 0);
87
88			if (wr > 0)
89			{
90			ptr += wr;
91			len -= wr;
92			}
93			else if (errno == EAGAIN)
94			{
95			dSP;
96			PUSHMARK (SP);
97			XPUSHs (OURDATAPTR->corosocket);
98			PUTBACK;
99			call_sv ((SV *)writable, G_VOID \| G_DISCARD);
100			}
101			else if (ptr == (char *)p)
102			return -1;
103			else
104			break;
105			}
106
107			return ptr - (char *)p;
108			}
109
110			MODULE = Coro::Mysql PACKAGE = Coro::Mysql
111
112			BOOT:
113			{
114			readable = get_cv ("Coro::Mysql::readable", 0);
115			writable = get_cv ("Coro::Mysql::writable", 0);
116			}
117
118			PROTOTYPES: ENABLE
119
120			void
121			_patch (IV sock, int fd, SV *corosocket)
122			CODE:
123			{
124			MYSQL my = (MYSQL )sock;
125			Vio *vio = my->net.vio;
126			ourdata *our;
127
128			if (fd != my->net.fd)
129			croak ("DBD::mysql fd and libmysql disagree - library mismatch, unsupported transport or API changes?");
130
131			if (fd != vio->sd)
132			croak ("DBD::mysql fd and vio-sd disagree - library mismatch, unsupported transport or API changes?");
133
134			if (vio->write != vio_write)
135			croak ("vio.write has unexpected content - library mismatch, unsupported transport or API changes?");
136
137			if (vio->read != vio_read && vio->read != vio_read_buff)
138			croak ("vio.read has unexpected content - library mismatch, unsupported transport or API changes?");
139
140			Newz (0, our, 1, ourdata);
141			our->magic = CoMy_MAGIC;
142			our->corosocket = newSVsv (corosocket);
143
144			vio->desc [DESC_OFFSET - 1] = 0;
145			OURDATAPTR = our;
146
147			vio->write = our_write;
148			vio->read = our_read;
149			}
150
151			void
152			_unpatch (IV sock)
153			CODE:
154			{
155			MYSQL my = (MYSQL )sock;
156			Vio *vio = my->net.vio;
157			my_bool dummy;
158
159			if (vio->read != our_read)
160			croak ("vio.read has unexpected content during unpatch - wtf?");
161
162	root	1.2	if (vio->write != our_write)
163			croak ("vio.write has unexpected content during unpatch - wtf?");
164
165	root	1.1	SvREFCNT_dec (OURDATAPTR->corosocket);
166
167			Safefree (OURDATAPTR);
168
169			vio->read = vio_read;
170			vio->write = vio_write;
171			}
172
173
174