mono/metadata/tpool-poll.c

   1 /*
   2  * tpool-poll.c: poll related stuff
   3  *
   4  * Authors:
   5  *   Dietmar Maurer (dietmar@ximian.com)
   6  *   Gonzalo Paniagua Javier (gonzalo@ximian.com)
   7  *
   8  * Copyright 2001-2003 Ximian, Inc (http://www.ximian.com)
   9  * Copyright 2004-2011 Novell, Inc (http://www.novell.com)
  10  */
  11 #include <config.h>
  12 #include <glib.h>
  13 #include <errno.h>
  14
  15 #include <mono/metadata/mono-ptr-array.h>
  16 #include <mono/metadata/threadpool.h>
  17 #include <mono/metadata/threadpool-internals.h>
  18 #include <mono/utils/mono-semaphore.h>
  19 #include <mono/utils/mono-poll.h>
  20
  21 #define INIT_POLLFD(a, b, c) {(a)->fd = b; (a)->events = c; (a)->revents = 0;}
  22 struct _tp_poll_data {
  23         int pipe [2];
  24         MonoSemType new_sem;
  25         mono_pollfd newpfd;
  26 };
  27
  28 typedef struct _tp_poll_data tp_poll_data;
  29
  30 static void tp_poll_shutdown (gpointer event_data);
  31 static void tp_poll_modify (gpointer p, int fd, int operation, int events, gboolean is_new);
  32 static void tp_poll_wait (gpointer p);
  33
  34 gpointer tp_poll_init (SocketIOData *data);
  35
  36 gpointer
  37 tp_poll_init (SocketIOData *data)
  38 {
  39         tp_poll_data *result;
  40 #ifdef HOST_WIN32
  41         struct sockaddr_in client;
  42         struct sockaddr_in server;
  43         SOCKET srv;
  44         int len;
  45 #endif
  46
  47         result = g_new0 (tp_poll_data, 1);
  48 #ifndef HOST_WIN32
  49         if (pipe (result->pipe) != 0) {
  50                 int err = errno;
  51                 perror ("mono");
  52                 g_assert (err);
  53         }
  54 #else
  55         srv = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
  56         g_assert (srv != INVALID_SOCKET);
  57         result->pipe [1] = socket (AF_INET, SOCK_STREAM, IPPROTO_TCP);
  58         g_assert (result->pipe [1] != INVALID_SOCKET);
  59
  60         server.sin_family = AF_INET;
  61         server.sin_addr.s_addr = inet_addr ("127.0.0.1");
  62         server.sin_port = 0;
  63         if (bind (srv, (SOCKADDR *) &server, sizeof (struct sockaddr_in))) {
  64                 g_print ("%d\n", WSAGetLastError ());
  65                 g_assert (1 != 0);
  66         }
  67
  68         len = sizeof (server);
  69         getsockname (srv, (SOCKADDR *) &server, &len);
  70         listen (srv, 1);
  71         if (connect ((SOCKET) result->pipe [1], (SOCKADDR *) &server, sizeof (server)) == SOCKET_ERROR) {
  72                 g_print ("%d\n", WSAGetLastError ());
  73                 g_assert (1 != 0);
  74         }
  75         len = sizeof (client);
  76         result->pipe [0] = accept (srv, (SOCKADDR *) &client, &len);
  77         g_assert (result->pipe [0] != INVALID_SOCKET);
  78         closesocket (srv);
  79 #endif
  80         MONO_SEM_INIT (&result->new_sem, 1);
  81         data->shutdown = tp_poll_shutdown;
  82         data->modify = tp_poll_modify;
  83         data->wait = tp_poll_wait;
  84         return result;
  85 }
  86
  87 static void
  88 tp_poll_modify (gpointer p, int fd, int operation, int events, gboolean is_new)
  89 {
  90         SocketIOData *socket_io_data;
  91         tp_poll_data *data;
  92         char msg [1];
  93         int unused G_GNUC_UNUSED;
  94
  95         socket_io_data = p;
  96         data = socket_io_data->event_data;
  97
  98         mono_mutex_unlock (&socket_io_data->io_lock);
  99
 100         MONO_SEM_WAIT (&data->new_sem);
 101         INIT_POLLFD (&data->newpfd, GPOINTER_TO_INT (fd), events);
 102         *msg = (char) operation;
 103 #ifndef HOST_WIN32
 104         unused = write (data->pipe [1], msg, 1);
 105 #else
 106         unused = send ((SOCKET) data->pipe [1], msg, 1, 0);
 107 #endif
 108 }
 109
 110 static void
 111 tp_poll_shutdown (gpointer event_data)
 112 {
 113         tp_poll_data *data = event_data;
 114
 115 #ifdef HOST_WIN32
 116         closesocket (data->pipe [0]);
 117         closesocket (data->pipe [1]);
 118 #else
 119         if (data->pipe [0] > -1)
 120                 close (data->pipe [0]);
 121         if (data->pipe [1] > -1)
 122                 close (data->pipe [1]);
 123 #endif
 124         data->pipe [0] = -1;
 125         data->pipe [1] = -1;
 126         MONO_SEM_DESTROY (&data->new_sem);
 127 }
 128
 129 static int
 130 mark_bad_fds (mono_pollfd *pfds, int nfds)
 131 {
 132         int i, ret;
 133         mono_pollfd *pfd;
 134         int count = 0;
 135
 136         for (i = 0; i < nfds; i++) {
 137                 pfd = &pfds [i];
 138                 if (pfd->fd == -1)
 139                         continue;
 140
 141                 ret = mono_poll (pfd, 1, 0);
 142                 if (ret == -1 && errno == EBADF) {
 143                         pfd->revents |= MONO_POLLNVAL;
 144                         count++;
 145                 } else if (ret == 1) {
 146                         count++;
 147                 }
 148         }
 149
 150         return count;
 151 }
 152
 153 static void
 154 tp_poll_wait (gpointer p)
 155 {
 156 #if MONO_SMALL_CONFIG
 157 #define INITIAL_POLLFD_SIZE     128
 158 #else
 159 #define INITIAL_POLLFD_SIZE     1024
 160 #endif
 161 #define POLL_ERRORS (MONO_POLLERR | MONO_POLLHUP | MONO_POLLNVAL)
 162
 163 #ifdef DISABLE_SOCKETS
 164 #define socket_io_cleanup(x)
 165 #endif
 166         mono_pollfd *pfds;
 167         gint maxfd = 1;
 168         gint allocated;
 169         gint i;
 170         tp_poll_data *data;
 171         SocketIOData *socket_io_data = p;
 172         MonoPtrArray async_results;
 173         gint nresults;
 174
 175         data = socket_io_data->event_data;
 176         allocated = INITIAL_POLLFD_SIZE;
 177         pfds = g_new0 (mono_pollfd, allocated);
 178         mono_ptr_array_init (async_results, allocated * 2);
 179         INIT_POLLFD (pfds, data->pipe [0], MONO_POLLIN);
 180         for (i = 1; i < allocated; i++)
 181                 INIT_POLLFD (&pfds [i], -1, 0);
 182
 183         while (1) {
 184                 int nsock = 0;
 185                 mono_pollfd *pfd;
 186                 char one [1];
 187                 MonoMList *list;
 188                 MonoObject *ares;
 189
 190                 mono_gc_set_skip_thread (TRUE);
 191
 192                 do {
 193                         if (nsock == -1) {
 194                                 check_for_interruption_critical ();
 195                         }
 196
 197                         nsock = mono_poll (pfds, maxfd, -1);
 198                 } while (nsock == -1 && errno == EINTR);
 199
 200                 mono_gc_set_skip_thread (FALSE);
 201
 202                 /*
 203                  * Apart from EINTR, we only check EBADF, for the rest:
 204                  *  EINVAL: mono_poll() 'protects' us from descriptor
 205                  *      numbers above the limit if using select() by marking
 206                  *      then as MONO_POLLERR.  If a system poll() is being
 207                  *      used, the number of descriptor we're passing will not
 208                  *      be over sysconf(_SC_OPEN_MAX), as the error would have
 209                  *      happened when opening.
 210                  *
 211                  *  EFAULT: we own the memory pointed by pfds.
 212                  *  ENOMEM: we're doomed anyway
 213                  *
 214                  */
 215
 216                 if (nsock == -1 && errno == EBADF) {
 217                         pfds->revents = 0; /* Just in case... */
 218                         nsock = mark_bad_fds (pfds, maxfd);
 219                 }
 220
 221                 if ((pfds->revents & POLL_ERRORS) != 0) {
 222                         /* We're supposed to die now, as the pipe has been closed */
 223                         g_free (pfds);
 224                         mono_ptr_array_destroy (async_results);
 225                         socket_io_cleanup (socket_io_data);
 226                         return;
 227                 }
 228
 229                 /* Got a new socket */
 230                 if ((pfds->revents & MONO_POLLIN) != 0) {
 231                         int nread;
 232                         gboolean found = FALSE;
 233
 234                         for (i = 1; i < allocated; i++) {
 235                                 pfd = &pfds [i];
 236                                 if (pfd->fd == data->newpfd.fd) {
 237                                         found = TRUE;
 238                                         break;
 239                                 }
 240                         }
 241
 242                         if (!found) {
 243                                 for (i = 1; i < allocated; i++) {
 244                                         pfd = &pfds [i];
 245                                         if (pfd->fd == -1)
 246                                                 break;
 247                                 }
 248                         }
 249
 250                         if (i == allocated) {
 251                                 mono_pollfd *oldfd;
 252
 253                                 oldfd = pfds;
 254                                 i = allocated;
 255                                 allocated = allocated * 2;
 256                                 pfds = g_renew (mono_pollfd, oldfd, allocated);
 257                                 g_free (oldfd);
 258                                 for (; i < allocated; i++)
 259                                         INIT_POLLFD (&pfds [i], -1, 0);
 260                                 //async_results = g_renew (gpointer, async_results, allocated * 2);
 261                         }
 262 #ifndef HOST_WIN32
 263                         nread = read (data->pipe [0], one, 1);
 264 #else
 265                         nread = recv ((SOCKET) data->pipe [0], one, 1, 0);
 266 #endif
 267                         if (nread <= 0) {
 268                                 g_free (pfds);
 269                                 mono_ptr_array_destroy (async_results);
 270                                 return; /* we're closed */
 271                         }
 272
 273                         INIT_POLLFD (&pfds [i], data->newpfd.fd, data->newpfd.events);
 274                         memset (&data->newpfd, 0, sizeof (mono_pollfd));
 275                         MONO_SEM_POST (&data->new_sem);
 276                         if (i >= maxfd)
 277                                 maxfd = i + 1;
 278                         nsock--;
 279                 }
 280
 281                 if (nsock == 0)
 282                         continue;
 283
 284                 mono_mutex_lock (&socket_io_data->io_lock);
 285                 if (socket_io_data->inited == 3) {
 286                         g_free (pfds);
 287                         mono_ptr_array_destroy (async_results);
 288                         mono_mutex_unlock (&socket_io_data->io_lock);
 289                         return; /* cleanup called */
 290                 }
 291
 292                 nresults = 0;
 293                 mono_ptr_array_clear (async_results);
 294
 295                 for (i = 1; i < maxfd && nsock > 0; i++) {
 296                         pfd = &pfds [i];
 297                         if (pfd->fd == -1 || pfd->revents == 0)
 298                                 continue;
 299
 300                         nsock--;
 301                         list = mono_g_hash_table_lookup (socket_io_data->sock_to_state, GINT_TO_POINTER (pfd->fd));
 302                         if (list != NULL && (pfd->revents & (MONO_POLLIN | POLL_ERRORS)) != 0) {
 303                                 ares = get_io_event (&list, MONO_POLLIN);
 304                                 if (ares != NULL) {
 305                                         mono_ptr_array_append (async_results, ares);
 306                                         ++nresults;
 307                                 }
 308                         }
 309
 310                         if (list != NULL && (pfd->revents & (MONO_POLLOUT | POLL_ERRORS)) != 0) {
 311                                 ares = get_io_event (&list, MONO_POLLOUT);
 312                                 if (ares != NULL) {
 313                                         mono_ptr_array_append (async_results, ares);
 314                                         ++nresults;
 315                                 }
 316                         }
 317
 318                         if (list != NULL) {
 319                                 mono_g_hash_table_replace (socket_io_data->sock_to_state, GINT_TO_POINTER (pfd->fd), list);
 320                                 pfd->events = get_events_from_list (list);
 321                         } else {
 322                                 mono_g_hash_table_remove (socket_io_data->sock_to_state, GINT_TO_POINTER (pfd->fd));
 323                                 pfd->fd = -1;
 324                                 if (i == maxfd - 1)
 325                                         maxfd--;
 326                         }
 327                 }
 328                 mono_mutex_unlock (&socket_io_data->io_lock);
 329                 threadpool_append_async_io_jobs ((MonoObject **) async_results.data, nresults);
 330                 mono_ptr_array_clear (async_results);
 331         }
 332 }
 333