Added suppor /and fix things for SunOS port
git-svn-id: https://svn.pjsip.org/repos/pjproject/main@2 74dad513-b988-da41-8d7b-12977e46ad98
diff --git a/pjlib/src/pj/ioqueue_winnt.c b/pjlib/src/pj/ioqueue_winnt.c
index 93116c9..2a8fd7e 100644
--- a/pjlib/src/pj/ioqueue_winnt.c
+++ b/pjlib/src/pj/ioqueue_winnt.c
@@ -1,852 +1,852 @@
-/* $Header: /pjproject-0.3/pjlib/src/pj/ioqueue_winnt.c 11 10/29/05 11:31a Bennylp $ */
-/* $Log: /pjproject-0.3/pjlib/src/pj/ioqueue_winnt.c $
- *
- * 11 10/29/05 11:31a Bennylp
- * Changed accept and lock.
- *
- * 10 10/14/05 12:26a Bennylp
- * Finished error code framework, some fixes in ioqueue, etc. Pretty
- * major.
- *
- * 9 9/17/05 10:37a Bennylp
- * Major reorganization towards version 0.3.
- *
- */
-#include <pj/ioqueue.h>
-#include <pj/os.h>
-#include <pj/lock.h>
-#include <pj/pool.h>
-#include <pj/string.h>
-#include <pj/sock.h>
-#include <pj/array.h>
-#include <pj/log.h>
-#include <pj/assert.h>
-#include <pj/errno.h>
-
-
-#if defined(PJ_HAS_WINSOCK2_H) && PJ_HAS_WINSOCK2_H != 0
-# include <winsock2.h>
-#elif defined(PJ_HAS_WINSOCK_H) && PJ_HAS_WINSOCK_H != 0
-# include <winsock.h>
-#endif
-
-#if defined(PJ_HAS_MSWSOCK_H) && PJ_HAS_MSWSOCK_H != 0
-# include <mswsock.h>
-#endif
-
-
-#define ACCEPT_ADDR_LEN (sizeof(pj_sockaddr_in)+20)
-
-/*
- * OVERLAP structure for send and receive.
- */
-typedef struct ioqueue_overlapped
-{
- WSAOVERLAPPED overlapped;
- pj_ioqueue_operation_e operation;
- WSABUF wsabuf;
-} ioqueue_overlapped;
-
-#if PJ_HAS_TCP
-/*
- * OVERLAP structure for accept.
- */
-typedef struct ioqueue_accept_rec
-{
- WSAOVERLAPPED overlapped;
- pj_ioqueue_operation_e operation;
- pj_sock_t newsock;
- pj_sock_t *newsock_ptr;
- int *addrlen;
- void *remote;
- void *local;
- char accept_buf[2 * ACCEPT_ADDR_LEN];
-} ioqueue_accept_rec;
-#endif
-
-/*
- * Structure for individual socket.
- */
-struct pj_ioqueue_key_t
-{
- HANDLE hnd;
- void *user_data;
- ioqueue_overlapped recv_overlapped;
- ioqueue_overlapped send_overlapped;
-#if PJ_HAS_TCP
- int connecting;
- ioqueue_accept_rec accept_overlapped;
-#endif
- pj_ioqueue_callback cb;
-};
-
-/*
- * IO Queue structure.
- */
-struct pj_ioqueue_t
-{
- HANDLE iocp;
- pj_lock_t *lock;
- pj_bool_t auto_delete_lock;
- unsigned event_count;
- HANDLE event_pool[MAXIMUM_WAIT_OBJECTS+1];
-#if PJ_HAS_TCP
- unsigned connecting_count;
- HANDLE connecting_handles[MAXIMUM_WAIT_OBJECTS+1];
- pj_ioqueue_key_t *connecting_keys[MAXIMUM_WAIT_OBJECTS+1];
-#endif
-};
-
-
-#if PJ_HAS_TCP
-/*
- * Process the socket when the overlapped accept() completed.
- */
-static void ioqueue_on_accept_complete(ioqueue_accept_rec *accept_overlapped)
-{
- struct sockaddr *local;
- struct sockaddr *remote;
- int locallen, remotelen;
-
- PJ_CHECK_STACK();
-
- /* Operation complete immediately. */
- GetAcceptExSockaddrs( accept_overlapped->accept_buf,
- 0,
- ACCEPT_ADDR_LEN,
- ACCEPT_ADDR_LEN,
- &local,
- &locallen,
- &remote,
- &remotelen);
- pj_memcpy(accept_overlapped->local, local, locallen);
- pj_memcpy(accept_overlapped->remote, remote, locallen);
- *accept_overlapped->addrlen = locallen;
- if (accept_overlapped->newsock_ptr)
- *accept_overlapped->newsock_ptr = accept_overlapped->newsock;
- accept_overlapped->operation = 0;
- accept_overlapped->newsock = PJ_INVALID_SOCKET;
-}
-
-static void erase_connecting_socket( pj_ioqueue_t *ioqueue, unsigned pos)
-{
- pj_ioqueue_key_t *key = ioqueue->connecting_keys[pos];
- HANDLE hEvent = ioqueue->connecting_handles[pos];
- unsigned long optval;
-
- /* Remove key from array of connecting handles. */
- pj_array_erase(ioqueue->connecting_keys, sizeof(key),
- ioqueue->connecting_count, pos);
- pj_array_erase(ioqueue->connecting_handles, sizeof(HANDLE),
- ioqueue->connecting_count, pos);
- --ioqueue->connecting_count;
-
- /* Disassociate the socket from the event. */
- WSAEventSelect((pj_sock_t)key->hnd, hEvent, 0);
-
- /* Put event object to pool. */
- if (ioqueue->event_count < MAXIMUM_WAIT_OBJECTS) {
- ioqueue->event_pool[ioqueue->event_count++] = hEvent;
- } else {
- /* Shouldn't happen. There should be no more pending connections
- * than max.
- */
- pj_assert(0);
- CloseHandle(hEvent);
- }
-
- /* Set socket to blocking again. */
- optval = 0;
- if (ioctlsocket((pj_sock_t)key->hnd, FIONBIO, &optval) != 0) {
- DWORD dwStatus;
- dwStatus = WSAGetLastError();
- }
-}
-
-/*
- * Poll for the completion of non-blocking connect().
- * If there's a completion, the function return the key of the completed
- * socket, and 'result' argument contains the connect() result. If connect()
- * succeeded, 'result' will have value zero, otherwise will have the error
- * code.
- */
-static pj_ioqueue_key_t *check_connecting( pj_ioqueue_t *ioqueue,
- pj_ssize_t *connect_err )
-{
- pj_ioqueue_key_t *key = NULL;
-
- if (ioqueue->connecting_count) {
- DWORD result;
-
- pj_lock_acquire(ioqueue->lock);
- result = WaitForMultipleObjects(ioqueue->connecting_count,
- ioqueue->connecting_handles,
- FALSE, 0);
- if (result >= WAIT_OBJECT_0 &&
- result < WAIT_OBJECT_0+ioqueue->connecting_count)
- {
- WSANETWORKEVENTS net_events;
-
- /* Got completed connect(). */
- unsigned pos = result - WAIT_OBJECT_0;
- key = ioqueue->connecting_keys[pos];
-
- /* See whether connect has succeeded. */
- WSAEnumNetworkEvents((pj_sock_t)key->hnd,
- ioqueue->connecting_handles[pos],
- &net_events);
- *connect_err = net_events.iErrorCode[FD_CONNECT_BIT];
-
- /* Erase socket from pending connect. */
- erase_connecting_socket(ioqueue, pos);
- }
- pj_lock_release(ioqueue->lock);
- }
- return key;
-}
-#endif
-
-
-PJ_DEF(pj_status_t) pj_ioqueue_create( pj_pool_t *pool,
- pj_size_t max_fd,
- int max_threads,
- pj_ioqueue_t **ioqueue)
-{
- pj_ioqueue_t *ioq;
- pj_status_t rc;
-
- PJ_UNUSED_ARG(max_fd);
- PJ_ASSERT_RETURN(pool && ioqueue, PJ_EINVAL);
-
- ioq = pj_pool_zalloc(pool, sizeof(*ioq));
- ioq->iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, max_threads);
- if (ioq->iocp == NULL)
- return PJ_RETURN_OS_ERROR(GetLastError());
-
- rc = pj_lock_create_simple_mutex(pool, NULL, &ioq->lock);
- if (rc != PJ_SUCCESS) {
- CloseHandle(ioq->iocp);
- return rc;
- }
-
- ioq->auto_delete_lock = PJ_TRUE;
-
- *ioqueue = ioq;
-
- PJ_LOG(4, ("pjlib", "WinNT IOCP I/O Queue created (%p)", ioq));
- return PJ_SUCCESS;
-}
-
-PJ_DEF(pj_status_t) pj_ioqueue_destroy( pj_ioqueue_t *ioque )
-{
- unsigned i;
-
- PJ_CHECK_STACK();
- PJ_ASSERT_RETURN(ioque, PJ_EINVAL);
-
- /* Destroy events in the pool */
- for (i=0; i<ioque->event_count; ++i) {
- CloseHandle(ioque->event_pool[i]);
- }
- ioque->event_count = 0;
-
- if (ioque->auto_delete_lock)
- pj_lock_destroy(ioque->lock);
-
- if (CloseHandle(ioque->iocp) == TRUE)
- return PJ_SUCCESS;
- else
- return PJ_RETURN_OS_ERROR(GetLastError());
-}
-
-PJ_DEF(pj_status_t) pj_ioqueue_set_lock( pj_ioqueue_t *ioque,
- pj_lock_t *lock,
- pj_bool_t auto_delete )
-{
- PJ_ASSERT_RETURN(ioque && lock, PJ_EINVAL);
-
- if (ioque->auto_delete_lock) {
- pj_lock_destroy(ioque->lock);
- }
-
- ioque->lock = lock;
- ioque->auto_delete_lock = auto_delete;
-
- return PJ_SUCCESS;
-}
-
-PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
- pj_ioqueue_t *ioque,
- pj_sock_t hnd,
- void *user_data,
- const pj_ioqueue_callback *cb,
- pj_ioqueue_key_t **key )
-{
- HANDLE hioq;
- pj_ioqueue_key_t *rec;
-
- PJ_ASSERT_RETURN(pool && ioque && cb && key, PJ_EINVAL);
-
- rec = pj_pool_zalloc(pool, sizeof(pj_ioqueue_key_t));
- rec->hnd = (HANDLE)hnd;
- rec->user_data = user_data;
- pj_memcpy(&rec->cb, cb, sizeof(pj_ioqueue_callback));
-#if PJ_HAS_TCP
- rec->accept_overlapped.newsock = PJ_INVALID_SOCKET;
-#endif
- hioq = CreateIoCompletionPort((HANDLE)hnd, ioque->iocp, (DWORD)rec, 0);
- if (!hioq) {
- return PJ_RETURN_OS_ERROR(GetLastError());
- }
-
- *key = rec;
- return PJ_SUCCESS;
-}
-
-
-
-PJ_DEF(pj_status_t) pj_ioqueue_unregister( pj_ioqueue_t *ioque,
- pj_ioqueue_key_t *key )
-{
- PJ_ASSERT_RETURN(ioque && key, PJ_EINVAL);
-
-#if PJ_HAS_TCP
- if (key->connecting) {
- unsigned pos;
-
- /* Erase from connecting_handles */
- pj_lock_acquire(ioque->lock);
- for (pos=0; pos < ioque->connecting_count; ++pos) {
- if (ioque->connecting_keys[pos] == key) {
- erase_connecting_socket(ioque, pos);
- if (key->accept_overlapped.newsock_ptr) {
- /* ??? shouldn't it be newsock instead of newsock_ptr??? */
- closesocket(*key->accept_overlapped.newsock_ptr);
- }
- break;
- }
- }
- pj_lock_release(ioque->lock);
- key->connecting = 0;
- }
-#endif
- return PJ_SUCCESS;
-}
-
-PJ_DEF(void*) pj_ioqueue_get_user_data( pj_ioqueue_key_t *key )
-{
- PJ_ASSERT_RETURN(key, NULL);
- return key->user_data;
-}
-
-/*
- * Poll for events.
- */
-PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t *ioque, const pj_time_val *timeout)
-{
- DWORD dwMsec, dwBytesTransfered, dwKey;
- ioqueue_overlapped *ov;
- pj_ioqueue_key_t *key;
- pj_ssize_t size_status;
- BOOL rc;
-
- PJ_ASSERT_RETURN(ioque, -PJ_EINVAL);
-
- /* Check the connecting array. */
-#if PJ_HAS_TCP
- key = check_connecting(ioque, &size_status);
- if (key != NULL) {
- key->cb.on_connect_complete(key, (int)size_status);
- return 1;
- }
-#endif
-
- /* Calculate miliseconds timeout for GetQueuedCompletionStatus */
- dwMsec = timeout ? timeout->sec*1000 + timeout->msec : INFINITE;
-
- /* Poll for completion status. */
- rc = GetQueuedCompletionStatus(ioque->iocp, &dwBytesTransfered, &dwKey,
- (OVERLAPPED**)&ov, dwMsec);
-
- /* The return value is:
- * - nonzero if event was dequeued.
- * - zero and ov==NULL if no event was dequeued.
- * - zero and ov!=NULL if event for failed I/O was dequeued.
- */
- if (ov) {
- /* Event was dequeued for either successfull or failed I/O */
- key = (pj_ioqueue_key_t*)dwKey;
- size_status = dwBytesTransfered;
- switch (ov->operation) {
- case PJ_IOQUEUE_OP_READ:
- case PJ_IOQUEUE_OP_RECV:
- case PJ_IOQUEUE_OP_RECV_FROM:
- key->recv_overlapped.operation = 0;
- if (key->cb.on_read_complete)
- key->cb.on_read_complete(key, size_status);
- break;
- case PJ_IOQUEUE_OP_WRITE:
- case PJ_IOQUEUE_OP_SEND:
- case PJ_IOQUEUE_OP_SEND_TO:
- key->send_overlapped.operation = 0;
- if (key->cb.on_write_complete)
- key->cb.on_write_complete(key, size_status);
- break;
-#if PJ_HAS_TCP
- case PJ_IOQUEUE_OP_ACCEPT:
- /* special case for accept. */
- ioqueue_on_accept_complete((ioqueue_accept_rec*)ov);
- if (key->cb.on_accept_complete)
- key->cb.on_accept_complete(key, key->accept_overlapped.newsock,
- 0);
- break;
- case PJ_IOQUEUE_OP_CONNECT:
-#endif
- case PJ_IOQUEUE_OP_NONE:
- pj_assert(0);
- break;
- }
- return 1;
- }
-
- if (GetLastError()==WAIT_TIMEOUT) {
- /* Check the connecting array. */
-#if PJ_HAS_TCP
- key = check_connecting(ioque, &size_status);
- if (key != NULL) {
- key->cb.on_connect_complete(key, (int)size_status);
- return 1;
- }
-#endif
- return 0;
- }
- return -1;
-}
-
-/*
- * pj_ioqueue_read()
- *
- * Initiate overlapped ReadFile operation.
- */
-PJ_DEF(pj_status_t) pj_ioqueue_read( pj_ioqueue_t *ioque,
- pj_ioqueue_key_t *key,
- void *buffer,
- pj_size_t buflen)
-{
- BOOL rc;
- DWORD bytesRead;
-
- PJ_CHECK_STACK();
- PJ_UNUSED_ARG(ioque);
-
- if (key->recv_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
- pj_assert(!"Operation already pending for this descriptor");
- return PJ_EBUSY;
- }
-
- pj_memset(&key->recv_overlapped, 0, sizeof(key->recv_overlapped));
- key->recv_overlapped.operation = PJ_IOQUEUE_OP_READ;
-
- rc = ReadFile(key->hnd, buffer, buflen, &bytesRead,
- &key->recv_overlapped.overlapped);
- if (rc == FALSE) {
- DWORD dwStatus = GetLastError();
- if (dwStatus==ERROR_IO_PENDING)
- return PJ_EPENDING;
- else
- return PJ_STATUS_FROM_OS(dwStatus);
- } else {
- /*
- * This is workaround to a probable bug in Win2000 (probably NT too).
- * Even if 'rc' is TRUE, which indicates operation has completed,
- * GetQueuedCompletionStatus still will return the key.
- * So as work around, we always return PJ_EPENDING here.
- */
- return PJ_EPENDING;
- }
-}
-
-/*
- * pj_ioqueue_recv()
- *
- * Initiate overlapped WSARecv() operation.
- */
-PJ_DEF(pj_status_t) pj_ioqueue_recv( pj_ioqueue_t *ioque,
- pj_ioqueue_key_t *key,
- void *buffer,
- pj_size_t buflen,
- unsigned flags )
-{
- int rc;
- DWORD bytesRead;
- DWORD dwFlags = 0;
-
- PJ_CHECK_STACK();
- PJ_UNUSED_ARG(ioque);
-
- if (key->recv_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
- pj_assert(!"Operation already pending for this socket");
- return PJ_EBUSY;
- }
-
- pj_memset(&key->recv_overlapped, 0, sizeof(key->recv_overlapped));
- key->recv_overlapped.operation = PJ_IOQUEUE_OP_READ;
-
- key->recv_overlapped.wsabuf.buf = buffer;
- key->recv_overlapped.wsabuf.len = buflen;
-
- dwFlags = flags;
-
- rc = WSARecv((SOCKET)key->hnd, &key->recv_overlapped.wsabuf, 1,
- &bytesRead, &dwFlags,
- &key->recv_overlapped.overlapped, NULL);
- if (rc == SOCKET_ERROR) {
- DWORD dwStatus = WSAGetLastError();
- if (dwStatus==WSA_IO_PENDING)
- return PJ_EPENDING;
- else
- return PJ_STATUS_FROM_OS(dwStatus);
- } else {
- /* Must always return pending status.
- * See comments on pj_ioqueue_read
- * return bytesRead;
- */
- return PJ_EPENDING;
- }
-}
-
-/*
- * pj_ioqueue_recvfrom()
- *
- * Initiate overlapped RecvFrom() operation.
- */
-PJ_DEF(pj_status_t) pj_ioqueue_recvfrom( pj_ioqueue_t *ioque,
- pj_ioqueue_key_t *key,
- void *buffer,
- pj_size_t buflen,
- unsigned flags,
- pj_sockaddr_t *addr,
- int *addrlen)
-{
- BOOL rc;
- DWORD bytesRead;
- DWORD dwFlags;
-
- PJ_CHECK_STACK();
- PJ_UNUSED_ARG(ioque);
-
- if (key->recv_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
- pj_assert(!"Operation already pending for this socket");
- return PJ_EBUSY;
- }
-
- pj_memset(&key->recv_overlapped, 0, sizeof(key->recv_overlapped));
- key->recv_overlapped.operation = PJ_IOQUEUE_OP_RECV_FROM;
- key->recv_overlapped.wsabuf.buf = buffer;
- key->recv_overlapped.wsabuf.len = buflen;
- dwFlags = flags;
- rc = WSARecvFrom((SOCKET)key->hnd, &key->recv_overlapped.wsabuf, 1,
- &bytesRead, &dwFlags,
- addr, addrlen,
- &key->recv_overlapped.overlapped, NULL);
- if (rc == SOCKET_ERROR) {
- DWORD dwStatus = WSAGetLastError();
- if (dwStatus==WSA_IO_PENDING)
- return PJ_EPENDING;
- else
- return PJ_STATUS_FROM_OS(dwStatus);
- } else {
- /* Must always return pending status.
- * See comments on pj_ioqueue_read
- * return bytesRead;
- */
- return PJ_EPENDING;
- }
-}
-
-/*
- * pj_ioqueue_write()
- *
- * Initiate overlapped WriteFile() operation.
- */
-PJ_DEF(pj_status_t) pj_ioqueue_write( pj_ioqueue_t *ioque,
- pj_ioqueue_key_t *key,
- const void *data,
- pj_size_t datalen)
-{
- BOOL rc;
- DWORD bytesWritten;
-
- PJ_CHECK_STACK();
- PJ_UNUSED_ARG(ioque);
-
- if (key->send_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
- pj_assert(!"Operation already pending for this descriptor");
- return PJ_EBUSY;
- }
-
- pj_memset(&key->send_overlapped, 0, sizeof(key->send_overlapped));
- key->send_overlapped.operation = PJ_IOQUEUE_OP_WRITE;
- rc = WriteFile(key->hnd, data, datalen, &bytesWritten,
- &key->send_overlapped.overlapped);
-
- if (rc == FALSE) {
- DWORD dwStatus = GetLastError();
- if (dwStatus==ERROR_IO_PENDING)
- return PJ_EPENDING;
- else
- return PJ_STATUS_FROM_OS(dwStatus);
- } else {
- /* Must always return pending status.
- * See comments on pj_ioqueue_read
- * return bytesWritten;
- */
- return PJ_EPENDING;
- }
-}
-
-
-/*
- * pj_ioqueue_send()
- *
- * Initiate overlapped Send operation.
- */
-PJ_DEF(pj_status_t) pj_ioqueue_send( pj_ioqueue_t *ioque,
- pj_ioqueue_key_t *key,
- const void *data,
- pj_size_t datalen,
- unsigned flags )
-{
- int rc;
- DWORD bytesWritten;
- DWORD dwFlags;
-
- PJ_CHECK_STACK();
- PJ_UNUSED_ARG(ioque);
-
- if (key->send_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
- pj_assert(!"Operation already pending for this socket");
- return PJ_EBUSY;
- }
-
- pj_memset(&key->send_overlapped, 0, sizeof(key->send_overlapped));
- key->send_overlapped.operation = PJ_IOQUEUE_OP_WRITE;
- key->send_overlapped.wsabuf.buf = (void*)data;
- key->send_overlapped.wsabuf.len = datalen;
- dwFlags = flags;
- rc = WSASend((SOCKET)key->hnd, &key->send_overlapped.wsabuf, 1,
- &bytesWritten, dwFlags,
- &key->send_overlapped.overlapped, NULL);
- if (rc == SOCKET_ERROR) {
- DWORD dwStatus = WSAGetLastError();
- if (dwStatus==WSA_IO_PENDING)
- return PJ_EPENDING;
- else
- return PJ_STATUS_FROM_OS(dwStatus);
- } else {
- /* Must always return pending status.
- * See comments on pj_ioqueue_read
- * return bytesRead;
- */
- return PJ_EPENDING;
- }
-}
-
-
-/*
- * pj_ioqueue_sendto()
- *
- * Initiate overlapped SendTo operation.
- */
-PJ_DEF(pj_status_t) pj_ioqueue_sendto( pj_ioqueue_t *ioque,
- pj_ioqueue_key_t *key,
- const void *data,
- pj_size_t datalen,
- unsigned flags,
- const pj_sockaddr_t *addr,
- int addrlen)
-{
- BOOL rc;
- DWORD bytesSent;
- DWORD dwFlags;
-
- PJ_CHECK_STACK();
- PJ_UNUSED_ARG(ioque);
-
- if (key->send_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
- pj_assert(!"Operation already pending for this socket");
- return PJ_EBUSY;
- }
-
- pj_memset(&key->send_overlapped, 0, sizeof(key->send_overlapped));
- key->send_overlapped.operation = PJ_IOQUEUE_OP_SEND_TO;
- key->send_overlapped.wsabuf.buf = (char*)data;
- key->send_overlapped.wsabuf.len = datalen;
- dwFlags = flags;
- rc = WSASendTo((SOCKET)key->hnd, &key->send_overlapped.wsabuf, 1,
- &bytesSent, dwFlags, addr,
- addrlen, &key->send_overlapped.overlapped, NULL);
- if (rc == SOCKET_ERROR) {
- DWORD dwStatus = WSAGetLastError();
- if (dwStatus==WSA_IO_PENDING)
- return PJ_EPENDING;
- else
- return PJ_STATUS_FROM_OS(dwStatus);
- } else {
- // Must always return pending status.
- // See comments on pj_ioqueue_read
- // return bytesSent;
- return PJ_EPENDING;
- }
-}
-
-#if PJ_HAS_TCP
-
-/*
- * pj_ioqueue_accept()
- *
- * Initiate overlapped accept() operation.
- */
-PJ_DEF(int) pj_ioqueue_accept( pj_ioqueue_t *ioqueue,
- pj_ioqueue_key_t *key,
- pj_sock_t *new_sock,
- pj_sockaddr_t *local,
- pj_sockaddr_t *remote,
- int *addrlen)
-{
- BOOL rc;
- DWORD bytesReceived;
- pj_status_t status;
-
- PJ_CHECK_STACK();
- PJ_UNUSED_ARG(ioqueue);
-
- if (key->accept_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
- pj_assert(!"Operation already pending for this socket");
- return PJ_EBUSY;
- }
-
- if (key->accept_overlapped.newsock == PJ_INVALID_SOCKET) {
- pj_sock_t sock;
- status = pj_sock_socket(PJ_AF_INET, PJ_SOCK_STREAM, 0, &sock);
- if (status != PJ_SUCCESS)
- return status;
-
- key->accept_overlapped.newsock = sock;
- }
- key->accept_overlapped.operation = PJ_IOQUEUE_OP_ACCEPT;
- key->accept_overlapped.addrlen = addrlen;
- key->accept_overlapped.local = local;
- key->accept_overlapped.remote = remote;
- key->accept_overlapped.newsock_ptr = new_sock;
- pj_memset(&key->accept_overlapped.overlapped, 0,
- sizeof(key->accept_overlapped.overlapped));
-
- rc = AcceptEx( (SOCKET)key->hnd, (SOCKET)key->accept_overlapped.newsock,
- key->accept_overlapped.accept_buf,
- 0, ACCEPT_ADDR_LEN, ACCEPT_ADDR_LEN,
- &bytesReceived,
- &key->accept_overlapped.overlapped);
-
- if (rc == TRUE) {
- ioqueue_on_accept_complete(&key->accept_overlapped);
- if (key->cb.on_accept_complete)
- key->cb.on_accept_complete(key, key->accept_overlapped.newsock, 0);
- return PJ_SUCCESS;
- } else {
- DWORD dwStatus = WSAGetLastError();
- if (dwStatus==WSA_IO_PENDING)
- return PJ_EPENDING;
- else
- return PJ_STATUS_FROM_OS(dwStatus);
- }
-}
-
-
-/*
- * pj_ioqueue_connect()
- *
- * Initiate overlapped connect() operation (well, it's non-blocking actually,
- * since there's no overlapped version of connect()).
- */
-PJ_DEF(pj_status_t) pj_ioqueue_connect( pj_ioqueue_t *ioqueue,
- pj_ioqueue_key_t *key,
- const pj_sockaddr_t *addr,
- int addrlen )
-{
- unsigned long optval = 1;
- HANDLE hEvent;
-
- PJ_CHECK_STACK();
-
- /* Set socket to non-blocking. */
- if (ioctlsocket((pj_sock_t)key->hnd, FIONBIO, &optval) != 0) {
- return PJ_RETURN_OS_ERROR(WSAGetLastError());
- }
-
- /* Initiate connect() */
- if (connect((pj_sock_t)key->hnd, addr, addrlen) != 0) {
- DWORD dwStatus;
- dwStatus = WSAGetLastError();
- if (dwStatus != WSAEWOULDBLOCK) {
- /* Permanent error */
- return PJ_RETURN_OS_ERROR(dwStatus);
- } else {
- /* Pending operation. This is what we're looking for. */
- }
- } else {
- /* Connect has completed immediately! */
- /* Restore to blocking mode. */
- optval = 0;
- if (ioctlsocket((pj_sock_t)key->hnd, FIONBIO, &optval) != 0) {
- return PJ_RETURN_OS_ERROR(WSAGetLastError());
- }
-
- key->cb.on_connect_complete(key, 0);
- return PJ_SUCCESS;
- }
-
- /* Add to the array of connecting socket to be polled */
- pj_lock_acquire(ioqueue->lock);
-
- if (ioqueue->connecting_count >= MAXIMUM_WAIT_OBJECTS) {
- pj_lock_release(ioqueue->lock);
- return PJ_ETOOMANYCONN;
- }
-
- /* Get or create event object. */
- if (ioqueue->event_count) {
- hEvent = ioqueue->event_pool[ioqueue->event_count - 1];
- --ioqueue->event_count;
- } else {
- hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
- if (hEvent == NULL) {
- DWORD dwStatus = GetLastError();
- pj_lock_release(ioqueue->lock);
- return PJ_STATUS_FROM_OS(dwStatus);
- }
- }
-
- /* Mark key as connecting.
- * We can't use array index since key can be removed dynamically.
- */
- key->connecting = 1;
-
- /* Associate socket events to the event object. */
- if (WSAEventSelect((pj_sock_t)key->hnd, hEvent, FD_CONNECT) != 0) {
- CloseHandle(hEvent);
- pj_lock_release(ioqueue->lock);
- return PJ_RETURN_OS_ERROR(WSAGetLastError());
- }
-
- /* Add to array. */
- ioqueue->connecting_keys[ ioqueue->connecting_count ] = key;
- ioqueue->connecting_handles[ ioqueue->connecting_count ] = hEvent;
- ioqueue->connecting_count++;
-
- pj_lock_release(ioqueue->lock);
-
- return PJ_EPENDING;
-}
-#endif /* #if PJ_HAS_TCP */
-
+/* $Header: /pjproject-0.3/pjlib/src/pj/ioqueue_winnt.c 11 10/29/05 11:31a Bennylp $ */
+/* $Log: /pjproject-0.3/pjlib/src/pj/ioqueue_winnt.c $
+ *
+ * 11 10/29/05 11:31a Bennylp
+ * Changed accept and lock.
+ *
+ * 10 10/14/05 12:26a Bennylp
+ * Finished error code framework, some fixes in ioqueue, etc. Pretty
+ * major.
+ *
+ * 9 9/17/05 10:37a Bennylp
+ * Major reorganization towards version 0.3.
+ *
+ */
+#include <pj/ioqueue.h>
+#include <pj/os.h>
+#include <pj/lock.h>
+#include <pj/pool.h>
+#include <pj/string.h>
+#include <pj/sock.h>
+#include <pj/array.h>
+#include <pj/log.h>
+#include <pj/assert.h>
+#include <pj/errno.h>
+
+
+#if defined(PJ_HAS_WINSOCK2_H) && PJ_HAS_WINSOCK2_H != 0
+# include <winsock2.h>
+#elif defined(PJ_HAS_WINSOCK_H) && PJ_HAS_WINSOCK_H != 0
+# include <winsock.h>
+#endif
+
+#if defined(PJ_HAS_MSWSOCK_H) && PJ_HAS_MSWSOCK_H != 0
+# include <mswsock.h>
+#endif
+
+
+#define ACCEPT_ADDR_LEN (sizeof(pj_sockaddr_in)+20)
+
+/*
+ * OVERLAP structure for send and receive.
+ */
+typedef struct ioqueue_overlapped
+{
+ WSAOVERLAPPED overlapped;
+ pj_ioqueue_operation_e operation;
+ WSABUF wsabuf;
+} ioqueue_overlapped;
+
+#if PJ_HAS_TCP
+/*
+ * OVERLAP structure for accept.
+ */
+typedef struct ioqueue_accept_rec
+{
+ WSAOVERLAPPED overlapped;
+ pj_ioqueue_operation_e operation;
+ pj_sock_t newsock;
+ pj_sock_t *newsock_ptr;
+ int *addrlen;
+ void *remote;
+ void *local;
+ char accept_buf[2 * ACCEPT_ADDR_LEN];
+} ioqueue_accept_rec;
+#endif
+
+/*
+ * Structure for individual socket.
+ */
+struct pj_ioqueue_key_t
+{
+ HANDLE hnd;
+ void *user_data;
+ ioqueue_overlapped recv_overlapped;
+ ioqueue_overlapped send_overlapped;
+#if PJ_HAS_TCP
+ int connecting;
+ ioqueue_accept_rec accept_overlapped;
+#endif
+ pj_ioqueue_callback cb;
+};
+
+/*
+ * IO Queue structure.
+ */
+struct pj_ioqueue_t
+{
+ HANDLE iocp;
+ pj_lock_t *lock;
+ pj_bool_t auto_delete_lock;
+ unsigned event_count;
+ HANDLE event_pool[MAXIMUM_WAIT_OBJECTS+1];
+#if PJ_HAS_TCP
+ unsigned connecting_count;
+ HANDLE connecting_handles[MAXIMUM_WAIT_OBJECTS+1];
+ pj_ioqueue_key_t *connecting_keys[MAXIMUM_WAIT_OBJECTS+1];
+#endif
+};
+
+
+#if PJ_HAS_TCP
+/*
+ * Process the socket when the overlapped accept() completed.
+ */
+static void ioqueue_on_accept_complete(ioqueue_accept_rec *accept_overlapped)
+{
+ struct sockaddr *local;
+ struct sockaddr *remote;
+ int locallen, remotelen;
+
+ PJ_CHECK_STACK();
+
+ /* Operation complete immediately. */
+ GetAcceptExSockaddrs( accept_overlapped->accept_buf,
+ 0,
+ ACCEPT_ADDR_LEN,
+ ACCEPT_ADDR_LEN,
+ &local,
+ &locallen,
+ &remote,
+ &remotelen);
+ pj_memcpy(accept_overlapped->local, local, locallen);
+ pj_memcpy(accept_overlapped->remote, remote, locallen);
+ *accept_overlapped->addrlen = locallen;
+ if (accept_overlapped->newsock_ptr)
+ *accept_overlapped->newsock_ptr = accept_overlapped->newsock;
+ accept_overlapped->operation = 0;
+ accept_overlapped->newsock = PJ_INVALID_SOCKET;
+}
+
+static void erase_connecting_socket( pj_ioqueue_t *ioqueue, unsigned pos)
+{
+ pj_ioqueue_key_t *key = ioqueue->connecting_keys[pos];
+ HANDLE hEvent = ioqueue->connecting_handles[pos];
+ unsigned long optval;
+
+ /* Remove key from array of connecting handles. */
+ pj_array_erase(ioqueue->connecting_keys, sizeof(key),
+ ioqueue->connecting_count, pos);
+ pj_array_erase(ioqueue->connecting_handles, sizeof(HANDLE),
+ ioqueue->connecting_count, pos);
+ --ioqueue->connecting_count;
+
+ /* Disassociate the socket from the event. */
+ WSAEventSelect((pj_sock_t)key->hnd, hEvent, 0);
+
+ /* Put event object to pool. */
+ if (ioqueue->event_count < MAXIMUM_WAIT_OBJECTS) {
+ ioqueue->event_pool[ioqueue->event_count++] = hEvent;
+ } else {
+ /* Shouldn't happen. There should be no more pending connections
+ * than max.
+ */
+ pj_assert(0);
+ CloseHandle(hEvent);
+ }
+
+ /* Set socket to blocking again. */
+ optval = 0;
+ if (ioctlsocket((pj_sock_t)key->hnd, FIONBIO, &optval) != 0) {
+ DWORD dwStatus;
+ dwStatus = WSAGetLastError();
+ }
+}
+
+/*
+ * Poll for the completion of non-blocking connect().
+ * If there's a completion, the function return the key of the completed
+ * socket, and 'result' argument contains the connect() result. If connect()
+ * succeeded, 'result' will have value zero, otherwise will have the error
+ * code.
+ */
+static pj_ioqueue_key_t *check_connecting( pj_ioqueue_t *ioqueue,
+ pj_ssize_t *connect_err )
+{
+ pj_ioqueue_key_t *key = NULL;
+
+ if (ioqueue->connecting_count) {
+ DWORD result;
+
+ pj_lock_acquire(ioqueue->lock);
+ result = WaitForMultipleObjects(ioqueue->connecting_count,
+ ioqueue->connecting_handles,
+ FALSE, 0);
+ if (result >= WAIT_OBJECT_0 &&
+ result < WAIT_OBJECT_0+ioqueue->connecting_count)
+ {
+ WSANETWORKEVENTS net_events;
+
+ /* Got completed connect(). */
+ unsigned pos = result - WAIT_OBJECT_0;
+ key = ioqueue->connecting_keys[pos];
+
+ /* See whether connect has succeeded. */
+ WSAEnumNetworkEvents((pj_sock_t)key->hnd,
+ ioqueue->connecting_handles[pos],
+ &net_events);
+ *connect_err = net_events.iErrorCode[FD_CONNECT_BIT];
+
+ /* Erase socket from pending connect. */
+ erase_connecting_socket(ioqueue, pos);
+ }
+ pj_lock_release(ioqueue->lock);
+ }
+ return key;
+}
+#endif
+
+
+PJ_DEF(pj_status_t) pj_ioqueue_create( pj_pool_t *pool,
+ pj_size_t max_fd,
+ int max_threads,
+ pj_ioqueue_t **ioqueue)
+{
+ pj_ioqueue_t *ioq;
+ pj_status_t rc;
+
+ PJ_UNUSED_ARG(max_fd);
+ PJ_ASSERT_RETURN(pool && ioqueue, PJ_EINVAL);
+
+ ioq = pj_pool_zalloc(pool, sizeof(*ioq));
+ ioq->iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, max_threads);
+ if (ioq->iocp == NULL)
+ return PJ_RETURN_OS_ERROR(GetLastError());
+
+ rc = pj_lock_create_simple_mutex(pool, NULL, &ioq->lock);
+ if (rc != PJ_SUCCESS) {
+ CloseHandle(ioq->iocp);
+ return rc;
+ }
+
+ ioq->auto_delete_lock = PJ_TRUE;
+
+ *ioqueue = ioq;
+
+ PJ_LOG(4, ("pjlib", "WinNT IOCP I/O Queue created (%p)", ioq));
+ return PJ_SUCCESS;
+}
+
+PJ_DEF(pj_status_t) pj_ioqueue_destroy( pj_ioqueue_t *ioque )
+{
+ unsigned i;
+
+ PJ_CHECK_STACK();
+ PJ_ASSERT_RETURN(ioque, PJ_EINVAL);
+
+ /* Destroy events in the pool */
+ for (i=0; i<ioque->event_count; ++i) {
+ CloseHandle(ioque->event_pool[i]);
+ }
+ ioque->event_count = 0;
+
+ if (ioque->auto_delete_lock)
+ pj_lock_destroy(ioque->lock);
+
+ if (CloseHandle(ioque->iocp) == TRUE)
+ return PJ_SUCCESS;
+ else
+ return PJ_RETURN_OS_ERROR(GetLastError());
+}
+
+PJ_DEF(pj_status_t) pj_ioqueue_set_lock( pj_ioqueue_t *ioque,
+ pj_lock_t *lock,
+ pj_bool_t auto_delete )
+{
+ PJ_ASSERT_RETURN(ioque && lock, PJ_EINVAL);
+
+ if (ioque->auto_delete_lock) {
+ pj_lock_destroy(ioque->lock);
+ }
+
+ ioque->lock = lock;
+ ioque->auto_delete_lock = auto_delete;
+
+ return PJ_SUCCESS;
+}
+
+PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
+ pj_ioqueue_t *ioque,
+ pj_sock_t hnd,
+ void *user_data,
+ const pj_ioqueue_callback *cb,
+ pj_ioqueue_key_t **key )
+{
+ HANDLE hioq;
+ pj_ioqueue_key_t *rec;
+
+ PJ_ASSERT_RETURN(pool && ioque && cb && key, PJ_EINVAL);
+
+ rec = pj_pool_zalloc(pool, sizeof(pj_ioqueue_key_t));
+ rec->hnd = (HANDLE)hnd;
+ rec->user_data = user_data;
+ pj_memcpy(&rec->cb, cb, sizeof(pj_ioqueue_callback));
+#if PJ_HAS_TCP
+ rec->accept_overlapped.newsock = PJ_INVALID_SOCKET;
+#endif
+ hioq = CreateIoCompletionPort((HANDLE)hnd, ioque->iocp, (DWORD)rec, 0);
+ if (!hioq) {
+ return PJ_RETURN_OS_ERROR(GetLastError());
+ }
+
+ *key = rec;
+ return PJ_SUCCESS;
+}
+
+
+
+PJ_DEF(pj_status_t) pj_ioqueue_unregister( pj_ioqueue_t *ioque,
+ pj_ioqueue_key_t *key )
+{
+ PJ_ASSERT_RETURN(ioque && key, PJ_EINVAL);
+
+#if PJ_HAS_TCP
+ if (key->connecting) {
+ unsigned pos;
+
+ /* Erase from connecting_handles */
+ pj_lock_acquire(ioque->lock);
+ for (pos=0; pos < ioque->connecting_count; ++pos) {
+ if (ioque->connecting_keys[pos] == key) {
+ erase_connecting_socket(ioque, pos);
+ if (key->accept_overlapped.newsock_ptr) {
+ /* ??? shouldn't it be newsock instead of newsock_ptr??? */
+ closesocket(*key->accept_overlapped.newsock_ptr);
+ }
+ break;
+ }
+ }
+ pj_lock_release(ioque->lock);
+ key->connecting = 0;
+ }
+#endif
+ return PJ_SUCCESS;
+}
+
+PJ_DEF(void*) pj_ioqueue_get_user_data( pj_ioqueue_key_t *key )
+{
+ PJ_ASSERT_RETURN(key, NULL);
+ return key->user_data;
+}
+
+/*
+ * Poll for events.
+ */
+PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t *ioque, const pj_time_val *timeout)
+{
+ DWORD dwMsec, dwBytesTransfered, dwKey;
+ ioqueue_overlapped *ov;
+ pj_ioqueue_key_t *key;
+ pj_ssize_t size_status;
+ BOOL rc;
+
+ PJ_ASSERT_RETURN(ioque, -PJ_EINVAL);
+
+ /* Check the connecting array. */
+#if PJ_HAS_TCP
+ key = check_connecting(ioque, &size_status);
+ if (key != NULL) {
+ key->cb.on_connect_complete(key, (int)size_status);
+ return 1;
+ }
+#endif
+
+ /* Calculate miliseconds timeout for GetQueuedCompletionStatus */
+ dwMsec = timeout ? timeout->sec*1000 + timeout->msec : INFINITE;
+
+ /* Poll for completion status. */
+ rc = GetQueuedCompletionStatus(ioque->iocp, &dwBytesTransfered, &dwKey,
+ (OVERLAPPED**)&ov, dwMsec);
+
+ /* The return value is:
+ * - nonzero if event was dequeued.
+ * - zero and ov==NULL if no event was dequeued.
+ * - zero and ov!=NULL if event for failed I/O was dequeued.
+ */
+ if (ov) {
+ /* Event was dequeued for either successfull or failed I/O */
+ key = (pj_ioqueue_key_t*)dwKey;
+ size_status = dwBytesTransfered;
+ switch (ov->operation) {
+ case PJ_IOQUEUE_OP_READ:
+ case PJ_IOQUEUE_OP_RECV:
+ case PJ_IOQUEUE_OP_RECV_FROM:
+ key->recv_overlapped.operation = 0;
+ if (key->cb.on_read_complete)
+ key->cb.on_read_complete(key, size_status);
+ break;
+ case PJ_IOQUEUE_OP_WRITE:
+ case PJ_IOQUEUE_OP_SEND:
+ case PJ_IOQUEUE_OP_SEND_TO:
+ key->send_overlapped.operation = 0;
+ if (key->cb.on_write_complete)
+ key->cb.on_write_complete(key, size_status);
+ break;
+#if PJ_HAS_TCP
+ case PJ_IOQUEUE_OP_ACCEPT:
+ /* special case for accept. */
+ ioqueue_on_accept_complete((ioqueue_accept_rec*)ov);
+ if (key->cb.on_accept_complete)
+ key->cb.on_accept_complete(key, key->accept_overlapped.newsock,
+ 0);
+ break;
+ case PJ_IOQUEUE_OP_CONNECT:
+#endif
+ case PJ_IOQUEUE_OP_NONE:
+ pj_assert(0);
+ break;
+ }
+ return 1;
+ }
+
+ if (GetLastError()==WAIT_TIMEOUT) {
+ /* Check the connecting array. */
+#if PJ_HAS_TCP
+ key = check_connecting(ioque, &size_status);
+ if (key != NULL) {
+ key->cb.on_connect_complete(key, (int)size_status);
+ return 1;
+ }
+#endif
+ return 0;
+ }
+ return -1;
+}
+
+/*
+ * pj_ioqueue_read()
+ *
+ * Initiate overlapped ReadFile operation.
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_read( pj_ioqueue_t *ioque,
+ pj_ioqueue_key_t *key,
+ void *buffer,
+ pj_size_t buflen)
+{
+ BOOL rc;
+ DWORD bytesRead;
+
+ PJ_CHECK_STACK();
+ PJ_UNUSED_ARG(ioque);
+
+ if (key->recv_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
+ pj_assert(!"Operation already pending for this descriptor");
+ return PJ_EBUSY;
+ }
+
+ pj_memset(&key->recv_overlapped, 0, sizeof(key->recv_overlapped));
+ key->recv_overlapped.operation = PJ_IOQUEUE_OP_READ;
+
+ rc = ReadFile(key->hnd, buffer, buflen, &bytesRead,
+ &key->recv_overlapped.overlapped);
+ if (rc == FALSE) {
+ DWORD dwStatus = GetLastError();
+ if (dwStatus==ERROR_IO_PENDING)
+ return PJ_EPENDING;
+ else
+ return PJ_STATUS_FROM_OS(dwStatus);
+ } else {
+ /*
+ * This is workaround to a probable bug in Win2000 (probably NT too).
+ * Even if 'rc' is TRUE, which indicates operation has completed,
+ * GetQueuedCompletionStatus still will return the key.
+ * So as work around, we always return PJ_EPENDING here.
+ */
+ return PJ_EPENDING;
+ }
+}
+
+/*
+ * pj_ioqueue_recv()
+ *
+ * Initiate overlapped WSARecv() operation.
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_recv( pj_ioqueue_t *ioque,
+ pj_ioqueue_key_t *key,
+ void *buffer,
+ pj_size_t buflen,
+ unsigned flags )
+{
+ int rc;
+ DWORD bytesRead;
+ DWORD dwFlags = 0;
+
+ PJ_CHECK_STACK();
+ PJ_UNUSED_ARG(ioque);
+
+ if (key->recv_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
+ pj_assert(!"Operation already pending for this socket");
+ return PJ_EBUSY;
+ }
+
+ pj_memset(&key->recv_overlapped, 0, sizeof(key->recv_overlapped));
+ key->recv_overlapped.operation = PJ_IOQUEUE_OP_READ;
+
+ key->recv_overlapped.wsabuf.buf = buffer;
+ key->recv_overlapped.wsabuf.len = buflen;
+
+ dwFlags = flags;
+
+ rc = WSARecv((SOCKET)key->hnd, &key->recv_overlapped.wsabuf, 1,
+ &bytesRead, &dwFlags,
+ &key->recv_overlapped.overlapped, NULL);
+ if (rc == SOCKET_ERROR) {
+ DWORD dwStatus = WSAGetLastError();
+ if (dwStatus==WSA_IO_PENDING)
+ return PJ_EPENDING;
+ else
+ return PJ_STATUS_FROM_OS(dwStatus);
+ } else {
+ /* Must always return pending status.
+ * See comments on pj_ioqueue_read
+ * return bytesRead;
+ */
+ return PJ_EPENDING;
+ }
+}
+
+/*
+ * pj_ioqueue_recvfrom()
+ *
+ * Initiate overlapped RecvFrom() operation.
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_recvfrom( pj_ioqueue_t *ioque,
+ pj_ioqueue_key_t *key,
+ void *buffer,
+ pj_size_t buflen,
+ unsigned flags,
+ pj_sockaddr_t *addr,
+ int *addrlen)
+{
+ BOOL rc;
+ DWORD bytesRead;
+ DWORD dwFlags;
+
+ PJ_CHECK_STACK();
+ PJ_UNUSED_ARG(ioque);
+
+ if (key->recv_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
+ pj_assert(!"Operation already pending for this socket");
+ return PJ_EBUSY;
+ }
+
+ pj_memset(&key->recv_overlapped, 0, sizeof(key->recv_overlapped));
+ key->recv_overlapped.operation = PJ_IOQUEUE_OP_RECV_FROM;
+ key->recv_overlapped.wsabuf.buf = buffer;
+ key->recv_overlapped.wsabuf.len = buflen;
+ dwFlags = flags;
+ rc = WSARecvFrom((SOCKET)key->hnd, &key->recv_overlapped.wsabuf, 1,
+ &bytesRead, &dwFlags,
+ addr, addrlen,
+ &key->recv_overlapped.overlapped, NULL);
+ if (rc == SOCKET_ERROR) {
+ DWORD dwStatus = WSAGetLastError();
+ if (dwStatus==WSA_IO_PENDING)
+ return PJ_EPENDING;
+ else
+ return PJ_STATUS_FROM_OS(dwStatus);
+ } else {
+ /* Must always return pending status.
+ * See comments on pj_ioqueue_read
+ * return bytesRead;
+ */
+ return PJ_EPENDING;
+ }
+}
+
+/*
+ * pj_ioqueue_write()
+ *
+ * Initiate overlapped WriteFile() operation.
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_write( pj_ioqueue_t *ioque,
+ pj_ioqueue_key_t *key,
+ const void *data,
+ pj_size_t datalen)
+{
+ BOOL rc;
+ DWORD bytesWritten;
+
+ PJ_CHECK_STACK();
+ PJ_UNUSED_ARG(ioque);
+
+ if (key->send_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
+ pj_assert(!"Operation already pending for this descriptor");
+ return PJ_EBUSY;
+ }
+
+ pj_memset(&key->send_overlapped, 0, sizeof(key->send_overlapped));
+ key->send_overlapped.operation = PJ_IOQUEUE_OP_WRITE;
+ rc = WriteFile(key->hnd, data, datalen, &bytesWritten,
+ &key->send_overlapped.overlapped);
+
+ if (rc == FALSE) {
+ DWORD dwStatus = GetLastError();
+ if (dwStatus==ERROR_IO_PENDING)
+ return PJ_EPENDING;
+ else
+ return PJ_STATUS_FROM_OS(dwStatus);
+ } else {
+ /* Must always return pending status.
+ * See comments on pj_ioqueue_read
+ * return bytesWritten;
+ */
+ return PJ_EPENDING;
+ }
+}
+
+
+/*
+ * pj_ioqueue_send()
+ *
+ * Initiate overlapped Send operation.
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_send( pj_ioqueue_t *ioque,
+ pj_ioqueue_key_t *key,
+ const void *data,
+ pj_size_t datalen,
+ unsigned flags )
+{
+ int rc;
+ DWORD bytesWritten;
+ DWORD dwFlags;
+
+ PJ_CHECK_STACK();
+ PJ_UNUSED_ARG(ioque);
+
+ if (key->send_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
+ pj_assert(!"Operation already pending for this socket");
+ return PJ_EBUSY;
+ }
+
+ pj_memset(&key->send_overlapped, 0, sizeof(key->send_overlapped));
+ key->send_overlapped.operation = PJ_IOQUEUE_OP_WRITE;
+ key->send_overlapped.wsabuf.buf = (void*)data;
+ key->send_overlapped.wsabuf.len = datalen;
+ dwFlags = flags;
+ rc = WSASend((SOCKET)key->hnd, &key->send_overlapped.wsabuf, 1,
+ &bytesWritten, dwFlags,
+ &key->send_overlapped.overlapped, NULL);
+ if (rc == SOCKET_ERROR) {
+ DWORD dwStatus = WSAGetLastError();
+ if (dwStatus==WSA_IO_PENDING)
+ return PJ_EPENDING;
+ else
+ return PJ_STATUS_FROM_OS(dwStatus);
+ } else {
+ /* Must always return pending status.
+ * See comments on pj_ioqueue_read
+ * return bytesRead;
+ */
+ return PJ_EPENDING;
+ }
+}
+
+
+/*
+ * pj_ioqueue_sendto()
+ *
+ * Initiate overlapped SendTo operation.
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_sendto( pj_ioqueue_t *ioque,
+ pj_ioqueue_key_t *key,
+ const void *data,
+ pj_size_t datalen,
+ unsigned flags,
+ const pj_sockaddr_t *addr,
+ int addrlen)
+{
+ BOOL rc;
+ DWORD bytesSent;
+ DWORD dwFlags;
+
+ PJ_CHECK_STACK();
+ PJ_UNUSED_ARG(ioque);
+
+ if (key->send_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
+ pj_assert(!"Operation already pending for this socket");
+ return PJ_EBUSY;
+ }
+
+ pj_memset(&key->send_overlapped, 0, sizeof(key->send_overlapped));
+ key->send_overlapped.operation = PJ_IOQUEUE_OP_SEND_TO;
+ key->send_overlapped.wsabuf.buf = (char*)data;
+ key->send_overlapped.wsabuf.len = datalen;
+ dwFlags = flags;
+ rc = WSASendTo((SOCKET)key->hnd, &key->send_overlapped.wsabuf, 1,
+ &bytesSent, dwFlags, addr,
+ addrlen, &key->send_overlapped.overlapped, NULL);
+ if (rc == SOCKET_ERROR) {
+ DWORD dwStatus = WSAGetLastError();
+ if (dwStatus==WSA_IO_PENDING)
+ return PJ_EPENDING;
+ else
+ return PJ_STATUS_FROM_OS(dwStatus);
+ } else {
+ // Must always return pending status.
+ // See comments on pj_ioqueue_read
+ // return bytesSent;
+ return PJ_EPENDING;
+ }
+}
+
+#if PJ_HAS_TCP
+
+/*
+ * pj_ioqueue_accept()
+ *
+ * Initiate overlapped accept() operation.
+ */
+PJ_DEF(int) pj_ioqueue_accept( pj_ioqueue_t *ioqueue,
+ pj_ioqueue_key_t *key,
+ pj_sock_t *new_sock,
+ pj_sockaddr_t *local,
+ pj_sockaddr_t *remote,
+ int *addrlen)
+{
+ BOOL rc;
+ DWORD bytesReceived;
+ pj_status_t status;
+
+ PJ_CHECK_STACK();
+ PJ_UNUSED_ARG(ioqueue);
+
+ if (key->accept_overlapped.operation != PJ_IOQUEUE_OP_NONE) {
+ pj_assert(!"Operation already pending for this socket");
+ return PJ_EBUSY;
+ }
+
+ if (key->accept_overlapped.newsock == PJ_INVALID_SOCKET) {
+ pj_sock_t sock;
+ status = pj_sock_socket(PJ_AF_INET, PJ_SOCK_STREAM, 0, &sock);
+ if (status != PJ_SUCCESS)
+ return status;
+
+ key->accept_overlapped.newsock = sock;
+ }
+ key->accept_overlapped.operation = PJ_IOQUEUE_OP_ACCEPT;
+ key->accept_overlapped.addrlen = addrlen;
+ key->accept_overlapped.local = local;
+ key->accept_overlapped.remote = remote;
+ key->accept_overlapped.newsock_ptr = new_sock;
+ pj_memset(&key->accept_overlapped.overlapped, 0,
+ sizeof(key->accept_overlapped.overlapped));
+
+ rc = AcceptEx( (SOCKET)key->hnd, (SOCKET)key->accept_overlapped.newsock,
+ key->accept_overlapped.accept_buf,
+ 0, ACCEPT_ADDR_LEN, ACCEPT_ADDR_LEN,
+ &bytesReceived,
+ &key->accept_overlapped.overlapped);
+
+ if (rc == TRUE) {
+ ioqueue_on_accept_complete(&key->accept_overlapped);
+ if (key->cb.on_accept_complete)
+ key->cb.on_accept_complete(key, key->accept_overlapped.newsock, 0);
+ return PJ_SUCCESS;
+ } else {
+ DWORD dwStatus = WSAGetLastError();
+ if (dwStatus==WSA_IO_PENDING)
+ return PJ_EPENDING;
+ else
+ return PJ_STATUS_FROM_OS(dwStatus);
+ }
+}
+
+
+/*
+ * pj_ioqueue_connect()
+ *
+ * Initiate overlapped connect() operation (well, it's non-blocking actually,
+ * since there's no overlapped version of connect()).
+ */
+PJ_DEF(pj_status_t) pj_ioqueue_connect( pj_ioqueue_t *ioqueue,
+ pj_ioqueue_key_t *key,
+ const pj_sockaddr_t *addr,
+ int addrlen )
+{
+ unsigned long optval = 1;
+ HANDLE hEvent;
+
+ PJ_CHECK_STACK();
+
+ /* Set socket to non-blocking. */
+ if (ioctlsocket((pj_sock_t)key->hnd, FIONBIO, &optval) != 0) {
+ return PJ_RETURN_OS_ERROR(WSAGetLastError());
+ }
+
+ /* Initiate connect() */
+ if (connect((pj_sock_t)key->hnd, addr, addrlen) != 0) {
+ DWORD dwStatus;
+ dwStatus = WSAGetLastError();
+ if (dwStatus != WSAEWOULDBLOCK) {
+ /* Permanent error */
+ return PJ_RETURN_OS_ERROR(dwStatus);
+ } else {
+ /* Pending operation. This is what we're looking for. */
+ }
+ } else {
+ /* Connect has completed immediately! */
+ /* Restore to blocking mode. */
+ optval = 0;
+ if (ioctlsocket((pj_sock_t)key->hnd, FIONBIO, &optval) != 0) {
+ return PJ_RETURN_OS_ERROR(WSAGetLastError());
+ }
+
+ key->cb.on_connect_complete(key, 0);
+ return PJ_SUCCESS;
+ }
+
+ /* Add to the array of connecting socket to be polled */
+ pj_lock_acquire(ioqueue->lock);
+
+ if (ioqueue->connecting_count >= MAXIMUM_WAIT_OBJECTS) {
+ pj_lock_release(ioqueue->lock);
+ return PJ_ETOOMANYCONN;
+ }
+
+ /* Get or create event object. */
+ if (ioqueue->event_count) {
+ hEvent = ioqueue->event_pool[ioqueue->event_count - 1];
+ --ioqueue->event_count;
+ } else {
+ hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
+ if (hEvent == NULL) {
+ DWORD dwStatus = GetLastError();
+ pj_lock_release(ioqueue->lock);
+ return PJ_STATUS_FROM_OS(dwStatus);
+ }
+ }
+
+ /* Mark key as connecting.
+ * We can't use array index since key can be removed dynamically.
+ */
+ key->connecting = 1;
+
+ /* Associate socket events to the event object. */
+ if (WSAEventSelect((pj_sock_t)key->hnd, hEvent, FD_CONNECT) != 0) {
+ CloseHandle(hEvent);
+ pj_lock_release(ioqueue->lock);
+ return PJ_RETURN_OS_ERROR(WSAGetLastError());
+ }
+
+ /* Add to array. */
+ ioqueue->connecting_keys[ ioqueue->connecting_count ] = key;
+ ioqueue->connecting_handles[ ioqueue->connecting_count ] = hEvent;
+ ioqueue->connecting_count++;
+
+ pj_lock_release(ioqueue->lock);
+
+ return PJ_EPENDING;
+}
+#endif /* #if PJ_HAS_TCP */
+