← Back to branch summary

~ubuntu-branches/ubuntu/wily/sflphone/wily

« back to all changes in this revision

Viewing changes to daemon/libs/pjproject-2.0.1/pjlib/src/pj/ioqueue_winnt.c

  • Committer: Package Import Robot
  • Author(s): Mark Purcell
  • Date: 2014-01-28 18:23:36 UTC
  • mfrom: (1.1.11)
  • mto: This revision was merged to the branch mainline in revision 24.
  • Revision ID: package-import@ubuntu.com-20140128182336-3xenud1kbnwmf3mz
http://bugs.debian.org/735846
http://bugs.debian.org/727164
http://bugs.debian.org/718178
http://bugs.debian.org/736583
http://bugs.debian.org/722040
http://bugs.debian.org/736746
* New upstream release 
  - Fixes "New Upstream Release" (Closes: #735846)
  - Fixes "Ringtone does not stop" (Closes: #727164)
  - Fixes "[sflphone-kde] crash on startup" (Closes: #718178)
  - Fixes "sflphone GUI crashes when call is hung up" (Closes: #736583)
* Build-Depends: ensure GnuTLS 2.6
  - libucommon-dev (>= 6.0.7-1.1), libccrtp-dev (>= 2.0.6-3)
  - Fixes "FTBFS Build-Depends libgnutls{26,28}-dev" (Closes: #722040)
* Fix "boost 1.49 is going away" unversioned Build-Depends: (Closes: #736746)
* Add Build-Depends: libsndfile-dev, nepomuk-core-dev

Show diffs side-by-side

added added

removed removed

Lines of Context:
daemon/libs/pjproject-2.0.1/pjlib/src/pj/ioqueue_winnt.c
1
 
/* $Id: ioqueue_winnt.c 3553 2011-05-05 06:14:19Z nanang $ */
2
 
/*
3
 
 * Copyright (C) 2008-2011 Teluu Inc. (http://www.teluu.com)
4
 
 * Copyright (C) 2003-2008 Benny Prijono <benny@prijono.org>
5
 
 *
6
 
 * This program is free software; you can redistribute it and/or modify
7
 
 * it under the terms of the GNU General Public License as published by
8
 
 * the Free Software Foundation; either version 2 of the License, or
9
 
 * (at your option) any later version.
10
 
 *
11
 
 * This program is distributed in the hope that it will be useful,
12
 
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
13
 
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
14
 
 * GNU General Public License for more details.
15
 
 *
16
 
 * You should have received a copy of the GNU General Public License
17
 
 * along with this program; if not, write to the Free Software
18
 
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
19
 
 */
20
 
#include <pj/ioqueue.h>
21
 
#include <pj/os.h>
22
 
#include <pj/lock.h>
23
 
#include <pj/pool.h>
24
 
#include <pj/string.h>
25
 
#include <pj/sock.h>
26
 
#include <pj/array.h>
27
 
#include <pj/log.h>
28
 
#include <pj/assert.h>
29
 
#include <pj/errno.h>
30
 
#include <pj/compat/socket.h>
31
 
 
32
 
 
33
 
#if defined(PJ_HAS_WINSOCK2_H) && PJ_HAS_WINSOCK2_H != 0
34
 
#  include <winsock2.h>
35
 
#elif defined(PJ_HAS_WINSOCK_H) && PJ_HAS_WINSOCK_H != 0
36
 
#  include <winsock.h>
37
 
#endif
38
 
 
39
 
#if defined(PJ_HAS_MSWSOCK_H) && PJ_HAS_MSWSOCK_H != 0
40
 
#  include <mswsock.h>
41
 
#endif
42
 
 
43
 
 
44
 
/* The address specified in AcceptEx() must be 16 more than the size of
45
 
 * SOCKADDR (source: MSDN).
46
 
 */
47
 
#define ACCEPT_ADDR_LEN     (sizeof(pj_sockaddr_in)+16)
48
 
 
49
 
typedef struct generic_overlapped
50
 
{
51
 
    WSAOVERLAPPED          overlapped;
52
 
    pj_ioqueue_operation_e operation;
53
 
} generic_overlapped;
54
 
 
55
 
/*
56
 
 * OVERLAPPPED structure for send and receive.
57
 
 */
58
 
typedef struct ioqueue_overlapped
59
 
{
60
 
    WSAOVERLAPPED          overlapped;
61
 
    pj_ioqueue_operation_e operation;
62
 
    WSABUF                 wsabuf;
63
 
    pj_sockaddr_in         dummy_addr;
64
 
    int                    dummy_addrlen;
65
 
} ioqueue_overlapped;
66
 
 
67
 
#if PJ_HAS_TCP
68
 
/*
69
 
 * OVERLAP structure for accept.
70
 
 */
71
 
typedef struct ioqueue_accept_rec
72
 
{
73
 
    WSAOVERLAPPED           overlapped;
74
 
    pj_ioqueue_operation_e  operation;
75
 
    pj_sock_t               newsock;
76
 
    pj_sock_t              *newsock_ptr;
77
 
    int                    *addrlen;
78
 
    void                   *remote;
79
 
    void                   *local;
80
 
    char                    accept_buf[2 * ACCEPT_ADDR_LEN];
81
 
} ioqueue_accept_rec;
82
 
#endif
83
 
 
84
 
/*
85
 
 * Structure to hold pending operation key.
86
 
 */
87
 
union operation_key
88
 
{
89
 
    generic_overlapped      generic;
90
 
    ioqueue_overlapped      overlapped;
91
 
#if PJ_HAS_TCP
92
 
    ioqueue_accept_rec      accept;
93
 
#endif
94
 
};
95
 
 
96
 
/* Type of handle in the key. */
97
 
enum handle_type
98
 
{
99
 
    HND_IS_UNKNOWN,
100
 
    HND_IS_FILE,
101
 
    HND_IS_SOCKET,
102
 
};
103
 
 
104
 
enum { POST_QUIT_LEN = 0xFFFFDEADUL };
105
 
 
106
 
/*
107
 
 * Structure for individual socket.
108
 
 */
109
 
struct pj_ioqueue_key_t
110
 
{
111
 
    PJ_DECL_LIST_MEMBER(struct pj_ioqueue_key_t);
112
 
 
113
 
    pj_ioqueue_t       *ioqueue;
114
 
    HANDLE              hnd;
115
 
    void               *user_data;
116
 
    enum handle_type    hnd_type;
117
 
    pj_ioqueue_callback cb;
118
 
    pj_bool_t           allow_concurrent;
119
 
 
120
 
#if PJ_HAS_TCP
121
 
    int                 connecting;
122
 
#endif
123
 
 
124
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
125
 
    pj_atomic_t        *ref_count;
126
 
    pj_bool_t           closing;
127
 
    pj_time_val         free_time;
128
 
    pj_mutex_t         *mutex;
129
 
#endif
130
 
 
131
 
};
132
 
 
133
 
/*
134
 
 * IO Queue structure.
135
 
 */
136
 
struct pj_ioqueue_t
137
 
{
138
 
    HANDLE            iocp;
139
 
    pj_lock_t        *lock;
140
 
    pj_bool_t         auto_delete_lock;
141
 
    pj_bool_t         default_concurrency;
142
 
 
143
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
144
 
    pj_ioqueue_key_t  active_list;
145
 
    pj_ioqueue_key_t  free_list;
146
 
    pj_ioqueue_key_t  closing_list;
147
 
#endif
148
 
 
149
 
    /* These are to keep track of connecting sockets */
150
 
#if PJ_HAS_TCP
151
 
    unsigned          event_count;
152
 
    HANDLE            event_pool[MAXIMUM_WAIT_OBJECTS+1];
153
 
    unsigned          connecting_count;
154
 
    HANDLE            connecting_handles[MAXIMUM_WAIT_OBJECTS+1];
155
 
    pj_ioqueue_key_t *connecting_keys[MAXIMUM_WAIT_OBJECTS+1];
156
 
#endif
157
 
};
158
 
 
159
 
 
160
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
161
 
/* Prototype */
162
 
static void scan_closing_keys(pj_ioqueue_t *ioqueue);
163
 
#endif
164
 
 
165
 
 
166
 
#if PJ_HAS_TCP
167
 
/*
168
 
 * Process the socket when the overlapped accept() completed.
169
 
 */
170
 
static void ioqueue_on_accept_complete(pj_ioqueue_key_t *key,
171
 
                                       ioqueue_accept_rec *accept_overlapped)
172
 
{
173
 
    struct sockaddr *local;
174
 
    struct sockaddr *remote;
175
 
    int locallen, remotelen;
176
 
    pj_status_t status;
177
 
 
178
 
    PJ_CHECK_STACK();
179
 
 
180
 
    /* On WinXP or later, use SO_UPDATE_ACCEPT_CONTEXT so that socket
181
 
     * addresses can be obtained with getsockname() and getpeername().
182
 
     */
183
 
    status = setsockopt(accept_overlapped->newsock, SOL_SOCKET,
184
 
                        SO_UPDATE_ACCEPT_CONTEXT,
185
 
                        (char*)&key->hnd,
186
 
                        sizeof(SOCKET));
187
 
    /* SO_UPDATE_ACCEPT_CONTEXT is for WinXP or later.
188
 
     * So ignore the error status.
189
 
     */
190
 
 
191
 
    /* Operation complete immediately. */
192
 
    if (accept_overlapped->addrlen) {
193
 
        GetAcceptExSockaddrs( accept_overlapped->accept_buf,
194
 
                              0,
195
 
                              ACCEPT_ADDR_LEN,
196
 
                              ACCEPT_ADDR_LEN,
197
 
                              &local,
198
 
                              &locallen,
199
 
                              &remote,
200
 
                              &remotelen);
201
 
        if (*accept_overlapped->addrlen >= locallen) {
202
 
            if (accept_overlapped->local)
203
 
                pj_memcpy(accept_overlapped->local, local, locallen);
204
 
            if (accept_overlapped->remote)
205
 
                pj_memcpy(accept_overlapped->remote, remote, locallen);
206
 
        } else {
207
 
            if (accept_overlapped->local)
208
 
                pj_bzero(accept_overlapped->local,
209
 
                         *accept_overlapped->addrlen);
210
 
            if (accept_overlapped->remote)
211
 
                pj_bzero(accept_overlapped->remote,
212
 
                         *accept_overlapped->addrlen);
213
 
        }
214
 
 
215
 
        *accept_overlapped->addrlen = locallen;
216
 
    }
217
 
    if (accept_overlapped->newsock_ptr)
218
 
        *accept_overlapped->newsock_ptr = accept_overlapped->newsock;
219
 
    accept_overlapped->operation = 0;
220
 
}
221
 
 
222
 
static void erase_connecting_socket( pj_ioqueue_t *ioqueue, unsigned pos)
223
 
{
224
 
    pj_ioqueue_key_t *key = ioqueue->connecting_keys[pos];
225
 
    HANDLE hEvent = ioqueue->connecting_handles[pos];
226
 
 
227
 
    /* Remove key from array of connecting handles. */
228
 
    pj_array_erase(ioqueue->connecting_keys, sizeof(key),
229
 
                   ioqueue->connecting_count, pos);
230
 
    pj_array_erase(ioqueue->connecting_handles, sizeof(HANDLE),
231
 
                   ioqueue->connecting_count, pos);
232
 
    --ioqueue->connecting_count;
233
 
 
234
 
    /* Disassociate the socket from the event. */
235
 
    WSAEventSelect((pj_sock_t)key->hnd, hEvent, 0);
236
 
 
237
 
    /* Put event object to pool. */
238
 
    if (ioqueue->event_count < MAXIMUM_WAIT_OBJECTS) {
239
 
        ioqueue->event_pool[ioqueue->event_count++] = hEvent;
240
 
    } else {
241
 
        /* Shouldn't happen. There should be no more pending connections
242
 
         * than max.
243
 
         */
244
 
        pj_assert(0);
245
 
        CloseHandle(hEvent);
246
 
    }
247
 
 
248
 
}
249
 
 
250
 
/*
251
 
 * Poll for the completion of non-blocking connect().
252
 
 * If there's a completion, the function return the key of the completed
253
 
 * socket, and 'result' argument contains the connect() result. If connect()
254
 
 * succeeded, 'result' will have value zero, otherwise will have the error
255
 
 * code.
256
 
 */
257
 
static int check_connecting( pj_ioqueue_t *ioqueue )
258
 
{
259
 
    if (ioqueue->connecting_count) {
260
 
        int i, count;
261
 
        struct
262
 
        {
263
 
            pj_ioqueue_key_t *key;
264
 
            pj_status_t       status;
265
 
        } events[PJ_IOQUEUE_MAX_EVENTS_IN_SINGLE_POLL-1];
266
 
 
267
 
        pj_lock_acquire(ioqueue->lock);
268
 
        for (count=0; count<PJ_IOQUEUE_MAX_EVENTS_IN_SINGLE_POLL-1; ++count) {
269
 
            DWORD result;
270
 
 
271
 
            result = WaitForMultipleObjects(ioqueue->connecting_count,
272
 
                                            ioqueue->connecting_handles,
273
 
                                            FALSE, 0);
274
 
            if (result >= WAIT_OBJECT_0 &&
275
 
                result < WAIT_OBJECT_0+ioqueue->connecting_count)
276
 
            {
277
 
                WSANETWORKEVENTS net_events;
278
 
 
279
 
                /* Got completed connect(). */
280
 
                unsigned pos = result - WAIT_OBJECT_0;
281
 
                events[count].key = ioqueue->connecting_keys[pos];
282
 
 
283
 
                /* See whether connect has succeeded. */
284
 
                WSAEnumNetworkEvents((pj_sock_t)events[count].key->hnd,
285
 
                                     ioqueue->connecting_handles[pos],
286
 
                                     &net_events);
287
 
                events[count].status =
288
 
                    PJ_STATUS_FROM_OS(net_events.iErrorCode[FD_CONNECT_BIT]);
289
 
 
290
 
                /* Erase socket from pending connect. */
291
 
                erase_connecting_socket(ioqueue, pos);
292
 
            } else {
293
 
                /* No more events */
294
 
                break;
295
 
            }
296
 
        }
297
 
        pj_lock_release(ioqueue->lock);
298
 
 
299
 
        /* Call callbacks. */
300
 
        for (i=0; i<count; ++i) {
301
 
            if (events[i].key->cb.on_connect_complete) {
302
 
                events[i].key->cb.on_connect_complete(events[i].key,
303
 
                                                      events[i].status);
304
 
            }
305
 
        }
306
 
 
307
 
        return count;
308
 
    }
309
 
 
310
 
    return 0;
311
 
 
312
 
}
313
 
#endif
314
 
 
315
 
/*
316
 
 * pj_ioqueue_name()
317
 
 */
318
 
PJ_DEF(const char*) pj_ioqueue_name(void)
319
 
{
320
 
    return "iocp";
321
 
}
322
 
 
323
 
/*
324
 
 * pj_ioqueue_create()
325
 
 */
326
 
PJ_DEF(pj_status_t) pj_ioqueue_create( pj_pool_t *pool,
327
 
                                       pj_size_t max_fd,
328
 
                                       pj_ioqueue_t **p_ioqueue)
329
 
{
330
 
    pj_ioqueue_t *ioqueue;
331
 
    unsigned i;
332
 
    pj_status_t rc;
333
 
 
334
 
    PJ_UNUSED_ARG(max_fd);
335
 
    PJ_ASSERT_RETURN(pool && p_ioqueue, PJ_EINVAL);
336
 
 
337
 
    rc = sizeof(union operation_key);
338
 
 
339
 
    /* Check that sizeof(pj_ioqueue_op_key_t) makes sense. */
340
 
    PJ_ASSERT_RETURN(sizeof(pj_ioqueue_op_key_t)-sizeof(void*) >=
341
 
                     sizeof(union operation_key), PJ_EBUG);
342
 
 
343
 
    /* Create IOCP */
344
 
    ioqueue = pj_pool_zalloc(pool, sizeof(*ioqueue));
345
 
    ioqueue->iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0);
346
 
    if (ioqueue->iocp == NULL)
347
 
        return PJ_RETURN_OS_ERROR(GetLastError());
348
 
 
349
 
    /* Create IOCP mutex */
350
 
    rc = pj_lock_create_recursive_mutex(pool, NULL, &ioqueue->lock);
351
 
    if (rc != PJ_SUCCESS) {
352
 
        CloseHandle(ioqueue->iocp);
353
 
        return rc;
354
 
    }
355
 
 
356
 
    ioqueue->auto_delete_lock = PJ_TRUE;
357
 
    ioqueue->default_concurrency = PJ_IOQUEUE_DEFAULT_ALLOW_CONCURRENCY;
358
 
 
359
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
360
 
    /*
361
 
     * Create and initialize key pools.
362
 
     */
363
 
    pj_list_init(&ioqueue->active_list);
364
 
    pj_list_init(&ioqueue->free_list);
365
 
    pj_list_init(&ioqueue->closing_list);
366
 
 
367
 
    /* Preallocate keys according to max_fd setting, and put them
368
 
     * in free_list.
369
 
     */
370
 
    for (i=0; i<max_fd; ++i) {
371
 
        pj_ioqueue_key_t *key;
372
 
 
373
 
        key = pj_pool_alloc(pool, sizeof(pj_ioqueue_key_t));
374
 
 
375
 
        rc = pj_atomic_create(pool, 0, &key->ref_count);
376
 
        if (rc != PJ_SUCCESS) {
377
 
            key = ioqueue->free_list.next;
378
 
            while (key != &ioqueue->free_list) {
379
 
                pj_atomic_destroy(key->ref_count);
380
 
                pj_mutex_destroy(key->mutex);
381
 
                key = key->next;
382
 
            }
383
 
            CloseHandle(ioqueue->iocp);
384
 
            return rc;
385
 
        }
386
 
 
387
 
        rc = pj_mutex_create_recursive(pool, "ioqkey", &key->mutex);
388
 
        if (rc != PJ_SUCCESS) {
389
 
            pj_atomic_destroy(key->ref_count);
390
 
            key = ioqueue->free_list.next;
391
 
            while (key != &ioqueue->free_list) {
392
 
                pj_atomic_destroy(key->ref_count);
393
 
                pj_mutex_destroy(key->mutex);
394
 
                key = key->next;
395
 
            }
396
 
            CloseHandle(ioqueue->iocp);
397
 
            return rc;
398
 
        }
399
 
 
400
 
        pj_list_push_back(&ioqueue->free_list, key);
401
 
    }
402
 
#endif
403
 
 
404
 
    *p_ioqueue = ioqueue;
405
 
 
406
 
    PJ_LOG(4, ("pjlib", "WinNT IOCP I/O Queue created (%p)", ioqueue));
407
 
    return PJ_SUCCESS;
408
 
}
409
 
 
410
 
/*
411
 
 * pj_ioqueue_destroy()
412
 
 */
413
 
PJ_DEF(pj_status_t) pj_ioqueue_destroy( pj_ioqueue_t *ioqueue )
414
 
{
415
 
#if PJ_HAS_TCP
416
 
    unsigned i;
417
 
#endif
418
 
    pj_ioqueue_key_t *key;
419
 
 
420
 
    PJ_CHECK_STACK();
421
 
    PJ_ASSERT_RETURN(ioqueue, PJ_EINVAL);
422
 
 
423
 
    pj_lock_acquire(ioqueue->lock);
424
 
 
425
 
#if PJ_HAS_TCP
426
 
    /* Destroy events in the pool */
427
 
    for (i=0; i<ioqueue->event_count; ++i) {
428
 
        CloseHandle(ioqueue->event_pool[i]);
429
 
    }
430
 
    ioqueue->event_count = 0;
431
 
#endif
432
 
 
433
 
    if (CloseHandle(ioqueue->iocp) != TRUE)
434
 
        return PJ_RETURN_OS_ERROR(GetLastError());
435
 
 
436
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
437
 
    /* Destroy reference counters */
438
 
    key = ioqueue->active_list.next;
439
 
    while (key != &ioqueue->active_list) {
440
 
        pj_atomic_destroy(key->ref_count);
441
 
        pj_mutex_destroy(key->mutex);
442
 
        key = key->next;
443
 
    }
444
 
 
445
 
    key = ioqueue->closing_list.next;
446
 
    while (key != &ioqueue->closing_list) {
447
 
        pj_atomic_destroy(key->ref_count);
448
 
        pj_mutex_destroy(key->mutex);
449
 
        key = key->next;
450
 
    }
451
 
 
452
 
    key = ioqueue->free_list.next;
453
 
    while (key != &ioqueue->free_list) {
454
 
        pj_atomic_destroy(key->ref_count);
455
 
        pj_mutex_destroy(key->mutex);
456
 
        key = key->next;
457
 
    }
458
 
#endif
459
 
 
460
 
    if (ioqueue->auto_delete_lock)
461
 
        pj_lock_destroy(ioqueue->lock);
462
 
 
463
 
    return PJ_SUCCESS;
464
 
}
465
 
 
466
 
 
467
 
PJ_DEF(pj_status_t) pj_ioqueue_set_default_concurrency(pj_ioqueue_t *ioqueue,
468
 
                                                       pj_bool_t allow)
469
 
{
470
 
    PJ_ASSERT_RETURN(ioqueue != NULL, PJ_EINVAL);
471
 
    ioqueue->default_concurrency = allow;
472
 
    return PJ_SUCCESS;
473
 
}
474
 
 
475
 
/*
476
 
 * pj_ioqueue_set_lock()
477
 
 */
478
 
PJ_DEF(pj_status_t) pj_ioqueue_set_lock( pj_ioqueue_t *ioqueue,
479
 
                                         pj_lock_t *lock,
480
 
                                         pj_bool_t auto_delete )
481
 
{
482
 
    PJ_ASSERT_RETURN(ioqueue && lock, PJ_EINVAL);
483
 
 
484
 
    if (ioqueue->auto_delete_lock) {
485
 
        pj_lock_destroy(ioqueue->lock);
486
 
    }
487
 
 
488
 
    ioqueue->lock = lock;
489
 
    ioqueue->auto_delete_lock = auto_delete;
490
 
 
491
 
    return PJ_SUCCESS;
492
 
}
493
 
 
494
 
/*
495
 
 * pj_ioqueue_register_sock()
496
 
 */
497
 
PJ_DEF(pj_status_t) pj_ioqueue_register_sock( pj_pool_t *pool,
498
 
                                              pj_ioqueue_t *ioqueue,
499
 
                                              pj_sock_t sock,
500
 
                                              void *user_data,
501
 
                                              const pj_ioqueue_callback *cb,
502
 
                                              pj_ioqueue_key_t **key )
503
 
{
504
 
    HANDLE hioq;
505
 
    pj_ioqueue_key_t *rec;
506
 
    u_long value;
507
 
    int rc;
508
 
 
509
 
    PJ_ASSERT_RETURN(pool && ioqueue && cb && key, PJ_EINVAL);
510
 
 
511
 
    pj_lock_acquire(ioqueue->lock);
512
 
 
513
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
514
 
    /* Scan closing list first to release unused keys.
515
 
     * Must do this with lock acquired.
516
 
     */
517
 
    scan_closing_keys(ioqueue);
518
 
 
519
 
    /* If safe unregistration is used, then get the key record from
520
 
     * the free list.
521
 
     */
522
 
    if (pj_list_empty(&ioqueue->free_list)) {
523
 
        pj_lock_release(ioqueue->lock);
524
 
        return PJ_ETOOMANY;
525
 
    }
526
 
 
527
 
    rec = ioqueue->free_list.next;
528
 
    pj_list_erase(rec);
529
 
 
530
 
    /* Set initial reference count to 1 */
531
 
    pj_assert(pj_atomic_get(rec->ref_count) == 0);
532
 
    pj_atomic_inc(rec->ref_count);
533
 
 
534
 
    rec->closing = 0;
535
 
 
536
 
#else
537
 
    rec = pj_pool_zalloc(pool, sizeof(pj_ioqueue_key_t));
538
 
#endif
539
 
 
540
 
    /* Build the key for this socket. */
541
 
    rec->ioqueue = ioqueue;
542
 
    rec->hnd = (HANDLE)sock;
543
 
    rec->hnd_type = HND_IS_SOCKET;
544
 
    rec->user_data = user_data;
545
 
    pj_memcpy(&rec->cb, cb, sizeof(pj_ioqueue_callback));
546
 
 
547
 
    /* Set concurrency for this handle */
548
 
    rc = pj_ioqueue_set_concurrency(rec, ioqueue->default_concurrency);
549
 
    if (rc != PJ_SUCCESS) {
550
 
        pj_lock_release(ioqueue->lock);
551
 
        return rc;
552
 
    }
553
 
 
554
 
#if PJ_HAS_TCP
555
 
    rec->connecting = 0;
556
 
#endif
557
 
 
558
 
    /* Set socket to nonblocking. */
559
 
    value = 1;
560
 
    rc = ioctlsocket(sock, FIONBIO, &value);
561
 
    if (rc != 0) {
562
 
        pj_lock_release(ioqueue->lock);
563
 
        return PJ_RETURN_OS_ERROR(WSAGetLastError());
564
 
    }
565
 
 
566
 
    /* Associate with IOCP */
567
 
    hioq = CreateIoCompletionPort((HANDLE)sock, ioqueue->iocp, (DWORD)rec, 0);
568
 
    if (!hioq) {
569
 
        pj_lock_release(ioqueue->lock);
570
 
        return PJ_RETURN_OS_ERROR(GetLastError());
571
 
    }
572
 
 
573
 
    *key = rec;
574
 
 
575
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
576
 
    pj_list_push_back(&ioqueue->active_list, rec);
577
 
#endif
578
 
 
579
 
    pj_lock_release(ioqueue->lock);
580
 
 
581
 
    return PJ_SUCCESS;
582
 
}
583
 
 
584
 
 
585
 
/*
586
 
 * pj_ioqueue_get_user_data()
587
 
 */
588
 
PJ_DEF(void*) pj_ioqueue_get_user_data( pj_ioqueue_key_t *key )
589
 
{
590
 
    PJ_ASSERT_RETURN(key, NULL);
591
 
    return key->user_data;
592
 
}
593
 
 
594
 
/*
595
 
 * pj_ioqueue_set_user_data()
596
 
 */
597
 
PJ_DEF(pj_status_t) pj_ioqueue_set_user_data( pj_ioqueue_key_t *key,
598
 
                                              void *user_data,
599
 
                                              void **old_data )
600
 
{
601
 
    PJ_ASSERT_RETURN(key, PJ_EINVAL);
602
 
 
603
 
    if (old_data)
604
 
        *old_data = key->user_data;
605
 
 
606
 
    key->user_data = user_data;
607
 
    return PJ_SUCCESS;
608
 
}
609
 
 
610
 
 
611
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
612
 
/* Decrement the key's reference counter, and when the counter reach zero,
613
 
 * destroy the key.
614
 
 */
615
 
static void decrement_counter(pj_ioqueue_key_t *key)
616
 
{
617
 
    if (pj_atomic_dec_and_get(key->ref_count) == 0) {
618
 
 
619
 
        pj_lock_acquire(key->ioqueue->lock);
620
 
 
621
 
        pj_assert(key->closing == 1);
622
 
        pj_gettickcount(&key->free_time);
623
 
        key->free_time.msec += PJ_IOQUEUE_KEY_FREE_DELAY;
624
 
        pj_time_val_normalize(&key->free_time);
625
 
 
626
 
        pj_list_erase(key);
627
 
        pj_list_push_back(&key->ioqueue->closing_list, key);
628
 
 
629
 
        pj_lock_release(key->ioqueue->lock);
630
 
    }
631
 
}
632
 
#endif
633
 
 
634
 
/*
635
 
 * Poll the I/O Completion Port, execute callback,
636
 
 * and return the key and bytes transfered of the last operation.
637
 
 */
638
 
static pj_bool_t poll_iocp( HANDLE hIocp, DWORD dwTimeout,
639
 
                            pj_ssize_t *p_bytes, pj_ioqueue_key_t **p_key )
640
 
{
641
 
    DWORD dwBytesTransfered, dwKey;
642
 
    generic_overlapped *pOv;
643
 
    pj_ioqueue_key_t *key;
644
 
    pj_ssize_t size_status = -1;
645
 
    BOOL rcGetQueued;
646
 
 
647
 
    /* Poll for completion status. */
648
 
    rcGetQueued = GetQueuedCompletionStatus(hIocp, &dwBytesTransfered,
649
 
                                            &dwKey, (OVERLAPPED**)&pOv,
650
 
                                            dwTimeout);
651
 
 
652
 
    /* The return value is:
653
 
     * - nonzero if event was dequeued.
654
 
     * - zero and pOv==NULL if no event was dequeued.
655
 
     * - zero and pOv!=NULL if event for failed I/O was dequeued.
656
 
     */
657
 
    if (pOv) {
658
 
        pj_bool_t has_lock;
659
 
 
660
 
        /* Event was dequeued for either successfull or failed I/O */
661
 
        key = (pj_ioqueue_key_t*)dwKey;
662
 
        size_status = dwBytesTransfered;
663
 
 
664
 
        /* Report to caller regardless */
665
 
        if (p_bytes)
666
 
            *p_bytes = size_status;
667
 
        if (p_key)
668
 
            *p_key = key;
669
 
 
670
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
671
 
        /* We shouldn't call callbacks if key is quitting. */
672
 
        if (key->closing)
673
 
            return PJ_TRUE;
674
 
 
675
 
        /* If concurrency is disabled, lock the key
676
 
         * (and save the lock status to local var since app may change
677
 
         * concurrency setting while in the callback) */
678
 
        if (key->allow_concurrent == PJ_FALSE) {
679
 
            pj_mutex_lock(key->mutex);
680
 
            has_lock = PJ_TRUE;
681
 
        } else {
682
 
            has_lock = PJ_FALSE;
683
 
        }
684
 
 
685
 
        /* Now that we get the lock, check again that key is not closing */
686
 
        if (key->closing) {
687
 
            if (has_lock) {
688
 
                pj_mutex_unlock(key->mutex);
689
 
            }
690
 
            return PJ_TRUE;
691
 
        }
692
 
 
693
 
        /* Increment reference counter to prevent this key from being
694
 
         * deleted
695
 
         */
696
 
        pj_atomic_inc(key->ref_count);
697
 
#else
698
 
        PJ_UNUSED_ARG(has_lock);
699
 
#endif
700
 
 
701
 
        /* Carry out the callback */
702
 
        switch (pOv->operation) {
703
 
        case PJ_IOQUEUE_OP_READ:
704
 
        case PJ_IOQUEUE_OP_RECV:
705
 
        case PJ_IOQUEUE_OP_RECV_FROM:
706
 
            pOv->operation = 0;
707
 
            if (key->cb.on_read_complete)
708
 
                key->cb.on_read_complete(key, (pj_ioqueue_op_key_t*)pOv,
709
 
                                         size_status);
710
 
            break;
711
 
        case PJ_IOQUEUE_OP_WRITE:
712
 
        case PJ_IOQUEUE_OP_SEND:
713
 
        case PJ_IOQUEUE_OP_SEND_TO:
714
 
            pOv->operation = 0;
715
 
            if (key->cb.on_write_complete)
716
 
                key->cb.on_write_complete(key, (pj_ioqueue_op_key_t*)pOv,
717
 
                                                size_status);
718
 
            break;
719
 
#if PJ_HAS_TCP
720
 
        case PJ_IOQUEUE_OP_ACCEPT:
721
 
            /* special case for accept. */
722
 
            ioqueue_on_accept_complete(key, (ioqueue_accept_rec*)pOv);
723
 
            if (key->cb.on_accept_complete) {
724
 
                ioqueue_accept_rec *accept_rec = (ioqueue_accept_rec*)pOv;
725
 
                pj_status_t status = PJ_SUCCESS;
726
 
                pj_sock_t newsock;
727
 
 
728
 
                newsock = accept_rec->newsock;
729
 
                accept_rec->newsock = PJ_INVALID_SOCKET;
730
 
 
731
 
                if (newsock == PJ_INVALID_SOCKET) {
732
 
                    int dwError = WSAGetLastError();
733
 
                    if (dwError == 0) dwError = OSERR_ENOTCONN;
734
 
                    status = PJ_RETURN_OS_ERROR(dwError);
735
 
                }
736
 
 
737
 
                key->cb.on_accept_complete(key, (pj_ioqueue_op_key_t*)pOv,
738
 
                                           newsock, status);
739
 
 
740
 
            }
741
 
            break;
742
 
        case PJ_IOQUEUE_OP_CONNECT:
743
 
#endif
744
 
        case PJ_IOQUEUE_OP_NONE:
745
 
            pj_assert(0);
746
 
            break;
747
 
        }
748
 
 
749
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
750
 
        decrement_counter(key);
751
 
        if (has_lock)
752
 
            pj_mutex_unlock(key->mutex);
753
 
#endif
754
 
 
755
 
        return PJ_TRUE;
756
 
    }
757
 
 
758
 
    /* No event was queued. */
759
 
    return PJ_FALSE;
760
 
}
761
 
 
762
 
/*
763
 
 * pj_ioqueue_unregister()
764
 
 */
765
 
PJ_DEF(pj_status_t) pj_ioqueue_unregister( pj_ioqueue_key_t *key )
766
 
{
767
 
    unsigned i;
768
 
    pj_bool_t has_lock;
769
 
    enum { RETRY = 10 };
770
 
 
771
 
    PJ_ASSERT_RETURN(key, PJ_EINVAL);
772
 
 
773
 
#if PJ_HAS_TCP
774
 
    if (key->connecting) {
775
 
        unsigned pos;
776
 
        pj_ioqueue_t *ioqueue;
777
 
 
778
 
        ioqueue = key->ioqueue;
779
 
 
780
 
        /* Erase from connecting_handles */
781
 
        pj_lock_acquire(ioqueue->lock);
782
 
        for (pos=0; pos < ioqueue->connecting_count; ++pos) {
783
 
            if (ioqueue->connecting_keys[pos] == key) {
784
 
                erase_connecting_socket(ioqueue, pos);
785
 
                break;
786
 
            }
787
 
        }
788
 
        key->connecting = 0;
789
 
        pj_lock_release(ioqueue->lock);
790
 
    }
791
 
#endif
792
 
 
793
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
794
 
    /* Mark key as closing before closing handle. */
795
 
    key->closing = 1;
796
 
 
797
 
    /* If concurrency is disabled, wait until the key has finished
798
 
     * processing the callback
799
 
     */
800
 
    if (key->allow_concurrent == PJ_FALSE) {
801
 
        pj_mutex_lock(key->mutex);
802
 
        has_lock = PJ_TRUE;
803
 
    } else {
804
 
        has_lock = PJ_FALSE;
805
 
    }
806
 
#else
807
 
    PJ_UNUSED_ARG(has_lock);
808
 
#endif
809
 
 
810
 
    /* Close handle (the only way to disassociate handle from IOCP).
811
 
     * We also need to close handle to make sure that no further events
812
 
     * will come to the handle.
813
 
     */
814
 
    /* Update 2008/07/18 (http://trac.pjsip.org/repos/ticket/575):
815
 
     *  - It seems that CloseHandle() in itself does not actually close
816
 
     *    the socket (i.e. it will still appear in "netstat" output). Also
817
 
     *    if we only use CloseHandle(), an "Invalid Handle" exception will
818
 
     *    be raised in WSACleanup().
819
 
     *  - MSDN documentation says that CloseHandle() must be called after
820
 
     *    closesocket() call (see
821
 
     *    http://msdn.microsoft.com/en-us/library/ms724211(VS.85).aspx).
822
 
     *    But turns out that this will raise "Invalid Handle" exception
823
 
     *    in debug mode.
824
 
     *  So because of this, we replaced CloseHandle() with closesocket()
825
 
     *  instead. These was tested on WinXP SP2.
826
 
     */
827
 
    //CloseHandle(key->hnd);
828
 
    pj_sock_close((pj_sock_t)key->hnd);
829
 
 
830
 
    /* Reset callbacks */
831
 
    key->cb.on_accept_complete = NULL;
832
 
    key->cb.on_connect_complete = NULL;
833
 
    key->cb.on_read_complete = NULL;
834
 
    key->cb.on_write_complete = NULL;
835
 
 
836
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
837
 
    /* Even after handle is closed, I suspect that IOCP may still try to
838
 
     * do something with the handle, causing memory corruption when pool
839
 
     * debugging is enabled.
840
 
     *
841
 
     * Forcing context switch seems to have fixed that, but this is quite
842
 
     * an ugly solution..
843
 
     *
844
 
     * Update 2008/02/13:
845
 
     *  This should not happen if concurrency is disallowed for the key.
846
 
     *  So at least application has a solution for this (i.e. by disallowing
847
 
     *  concurrency in the key).
848
 
     */
849
 
    //This will loop forever if unregistration is done on the callback.
850
 
    //Doing this with RETRY I think should solve the IOCP setting the
851
 
    //socket signalled, without causing the deadlock.
852
 
    //while (pj_atomic_get(key->ref_count) != 1)
853
 
    //  pj_thread_sleep(0);
854
 
    for (i=0; pj_atomic_get(key->ref_count) != 1 && i<RETRY; ++i)
855
 
        pj_thread_sleep(0);
856
 
 
857
 
    /* Decrement reference counter to destroy the key. */
858
 
    decrement_counter(key);
859
 
 
860
 
    if (has_lock)
861
 
        pj_mutex_unlock(key->mutex);
862
 
#endif
863
 
 
864
 
    return PJ_SUCCESS;
865
 
}
866
 
 
867
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
868
 
/* Scan the closing list, and put pending closing keys to free list.
869
 
 * Must do this with ioqueue mutex held.
870
 
 */
871
 
static void scan_closing_keys(pj_ioqueue_t *ioqueue)
872
 
{
873
 
    if (!pj_list_empty(&ioqueue->closing_list)) {
874
 
        pj_time_val now;
875
 
        pj_ioqueue_key_t *key;
876
 
 
877
 
        pj_gettickcount(&now);
878
 
 
879
 
        /* Move closing keys to free list when they've finished the closing
880
 
         * idle time.
881
 
         */
882
 
        key = ioqueue->closing_list.next;
883
 
        while (key != &ioqueue->closing_list) {
884
 
            pj_ioqueue_key_t *next = key->next;
885
 
 
886
 
            pj_assert(key->closing != 0);
887
 
 
888
 
            if (PJ_TIME_VAL_GTE(now, key->free_time)) {
889
 
                pj_list_erase(key);
890
 
                pj_list_push_back(&ioqueue->free_list, key);
891
 
            }
892
 
            key = next;
893
 
        }
894
 
    }
895
 
}
896
 
#endif
897
 
 
898
 
/*
899
 
 * pj_ioqueue_poll()
900
 
 *
901
 
 * Poll for events.
902
 
 */
903
 
PJ_DEF(int) pj_ioqueue_poll( pj_ioqueue_t *ioqueue, const pj_time_val *timeout)
904
 
{
905
 
    DWORD dwMsec;
906
 
#if PJ_HAS_TCP
907
 
    int connect_count = 0;
908
 
#endif
909
 
    int event_count = 0;
910
 
 
911
 
    PJ_ASSERT_RETURN(ioqueue, -PJ_EINVAL);
912
 
 
913
 
    /* Calculate miliseconds timeout for GetQueuedCompletionStatus */
914
 
    dwMsec = timeout ? timeout->sec*1000 + timeout->msec : INFINITE;
915
 
 
916
 
    /* Poll for completion status. */
917
 
    event_count = poll_iocp(ioqueue->iocp, dwMsec, NULL, NULL);
918
 
 
919
 
#if PJ_HAS_TCP
920
 
    /* Check the connecting array, only when there's no activity. */
921
 
    if (event_count == 0) {
922
 
        connect_count = check_connecting(ioqueue);
923
 
        if (connect_count > 0)
924
 
            event_count += connect_count;
925
 
    }
926
 
#endif
927
 
 
928
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
929
 
    /* Check the closing keys only when there's no activity and when there are
930
 
     * pending closing keys.
931
 
     */
932
 
    if (event_count == 0 && !pj_list_empty(&ioqueue->closing_list)) {
933
 
        pj_lock_acquire(ioqueue->lock);
934
 
        scan_closing_keys(ioqueue);
935
 
        pj_lock_release(ioqueue->lock);
936
 
    }
937
 
#endif
938
 
 
939
 
    /* Return number of events. */
940
 
    return event_count;
941
 
}
942
 
 
943
 
/*
944
 
 * pj_ioqueue_recv()
945
 
 *
946
 
 * Initiate overlapped WSARecv() operation.
947
 
 */
948
 
PJ_DEF(pj_status_t) pj_ioqueue_recv(  pj_ioqueue_key_t *key,
949
 
                                      pj_ioqueue_op_key_t *op_key,
950
 
                                      void *buffer,
951
 
                                      pj_ssize_t *length,
952
 
                                      pj_uint32_t flags )
953
 
{
954
 
    /*
955
 
     * Ideally we should just call pj_ioqueue_recvfrom() with NULL addr and
956
 
     * addrlen here. But unfortunately it generates EINVAL... :-(
957
 
     *  -bennylp
958
 
     */
959
 
    int rc;
960
 
    DWORD bytesRead;
961
 
    DWORD dwFlags = 0;
962
 
    union operation_key *op_key_rec;
963
 
 
964
 
    PJ_CHECK_STACK();
965
 
    PJ_ASSERT_RETURN(key && op_key && buffer && length, PJ_EINVAL);
966
 
 
967
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
968
 
    /* Check key is not closing */
969
 
    if (key->closing)
970
 
        return PJ_ECANCELLED;
971
 
#endif
972
 
 
973
 
    op_key_rec = (union operation_key*)op_key->internal__;
974
 
    op_key_rec->overlapped.wsabuf.buf = buffer;
975
 
    op_key_rec->overlapped.wsabuf.len = *length;
976
 
 
977
 
    dwFlags = flags;
978
 
 
979
 
    /* Try non-overlapped received first to see if data is
980
 
     * immediately available.
981
 
     */
982
 
    if ((flags & PJ_IOQUEUE_ALWAYS_ASYNC) == 0) {
983
 
        rc = WSARecv((SOCKET)key->hnd, &op_key_rec->overlapped.wsabuf, 1,
984
 
                     &bytesRead, &dwFlags, NULL, NULL);
985
 
        if (rc == 0) {
986
 
            *length = bytesRead;
987
 
            return PJ_SUCCESS;
988
 
        } else {
989
 
            DWORD dwError = WSAGetLastError();
990
 
            if (dwError != WSAEWOULDBLOCK) {
991
 
                *length = -1;
992
 
                return PJ_RETURN_OS_ERROR(dwError);
993
 
            }
994
 
        }
995
 
    }
996
 
 
997
 
    dwFlags &= ~(PJ_IOQUEUE_ALWAYS_ASYNC);
998
 
 
999
 
    /*
1000
 
     * No immediate data available.
1001
 
     * Register overlapped Recv() operation.
1002
 
     */
1003
 
    pj_bzero( &op_key_rec->overlapped.overlapped,
1004
 
              sizeof(op_key_rec->overlapped.overlapped));
1005
 
    op_key_rec->overlapped.operation = PJ_IOQUEUE_OP_RECV;
1006
 
 
1007
 
    rc = WSARecv((SOCKET)key->hnd, &op_key_rec->overlapped.wsabuf, 1,
1008
 
                  &bytesRead, &dwFlags,
1009
 
                  &op_key_rec->overlapped.overlapped, NULL);
1010
 
    if (rc == SOCKET_ERROR) {
1011
 
        DWORD dwStatus = WSAGetLastError();
1012
 
        if (dwStatus!=WSA_IO_PENDING) {
1013
 
            *length = -1;
1014
 
            return PJ_STATUS_FROM_OS(dwStatus);
1015
 
        }
1016
 
    }
1017
 
 
1018
 
    /* Pending operation has been scheduled. */
1019
 
    return PJ_EPENDING;
1020
 
}
1021
 
 
1022
 
/*
1023
 
 * pj_ioqueue_recvfrom()
1024
 
 *
1025
 
 * Initiate overlapped RecvFrom() operation.
1026
 
 */
1027
 
PJ_DEF(pj_status_t) pj_ioqueue_recvfrom( pj_ioqueue_key_t *key,
1028
 
                                         pj_ioqueue_op_key_t *op_key,
1029
 
                                         void *buffer,
1030
 
                                         pj_ssize_t *length,
1031
 
                                         pj_uint32_t flags,
1032
 
                                         pj_sockaddr_t *addr,
1033
 
                                         int *addrlen)
1034
 
{
1035
 
    int rc;
1036
 
    DWORD bytesRead;
1037
 
    DWORD dwFlags = 0;
1038
 
    union operation_key *op_key_rec;
1039
 
 
1040
 
    PJ_CHECK_STACK();
1041
 
    PJ_ASSERT_RETURN(key && op_key && buffer, PJ_EINVAL);
1042
 
 
1043
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
1044
 
    /* Check key is not closing */
1045
 
    if (key->closing)
1046
 
        return PJ_ECANCELLED;
1047
 
#endif
1048
 
 
1049
 
    op_key_rec = (union operation_key*)op_key->internal__;
1050
 
    op_key_rec->overlapped.wsabuf.buf = buffer;
1051
 
    op_key_rec->overlapped.wsabuf.len = *length;
1052
 
 
1053
 
    dwFlags = flags;
1054
 
 
1055
 
    /* Try non-overlapped received first to see if data is
1056
 
     * immediately available.
1057
 
     */
1058
 
    if ((flags & PJ_IOQUEUE_ALWAYS_ASYNC) == 0) {
1059
 
        rc = WSARecvFrom((SOCKET)key->hnd, &op_key_rec->overlapped.wsabuf, 1,
1060
 
                         &bytesRead, &dwFlags, addr, addrlen, NULL, NULL);
1061
 
        if (rc == 0) {
1062
 
            *length = bytesRead;
1063
 
            return PJ_SUCCESS;
1064
 
        } else {
1065
 
            DWORD dwError = WSAGetLastError();
1066
 
            if (dwError != WSAEWOULDBLOCK) {
1067
 
                *length = -1;
1068
 
                return PJ_RETURN_OS_ERROR(dwError);
1069
 
            }
1070
 
        }
1071
 
    }
1072
 
 
1073
 
    dwFlags &= ~(PJ_IOQUEUE_ALWAYS_ASYNC);
1074
 
 
1075
 
    /*
1076
 
     * No immediate data available.
1077
 
     * Register overlapped Recv() operation.
1078
 
     */
1079
 
    pj_bzero( &op_key_rec->overlapped.overlapped,
1080
 
              sizeof(op_key_rec->overlapped.overlapped));
1081
 
    op_key_rec->overlapped.operation = PJ_IOQUEUE_OP_RECV;
1082
 
 
1083
 
    rc = WSARecvFrom((SOCKET)key->hnd, &op_key_rec->overlapped.wsabuf, 1,
1084
 
                     &bytesRead, &dwFlags, addr, addrlen,
1085
 
                     &op_key_rec->overlapped.overlapped, NULL);
1086
 
    if (rc == SOCKET_ERROR) {
1087
 
        DWORD dwStatus = WSAGetLastError();
1088
 
        if (dwStatus!=WSA_IO_PENDING) {
1089
 
            *length = -1;
1090
 
            return PJ_STATUS_FROM_OS(dwStatus);
1091
 
        }
1092
 
    }
1093
 
 
1094
 
    /* Pending operation has been scheduled. */
1095
 
    return PJ_EPENDING;
1096
 
}
1097
 
 
1098
 
/*
1099
 
 * pj_ioqueue_send()
1100
 
 *
1101
 
 * Initiate overlapped Send operation.
1102
 
 */
1103
 
PJ_DEF(pj_status_t) pj_ioqueue_send(  pj_ioqueue_key_t *key,
1104
 
                                      pj_ioqueue_op_key_t *op_key,
1105
 
                                      const void *data,
1106
 
                                      pj_ssize_t *length,
1107
 
                                      pj_uint32_t flags )
1108
 
{
1109
 
    return pj_ioqueue_sendto(key, op_key, data, length, flags, NULL, 0);
1110
 
}
1111
 
 
1112
 
 
1113
 
/*
1114
 
 * pj_ioqueue_sendto()
1115
 
 *
1116
 
 * Initiate overlapped SendTo operation.
1117
 
 */
1118
 
PJ_DEF(pj_status_t) pj_ioqueue_sendto( pj_ioqueue_key_t *key,
1119
 
                                       pj_ioqueue_op_key_t *op_key,
1120
 
                                       const void *data,
1121
 
                                       pj_ssize_t *length,
1122
 
                                       pj_uint32_t flags,
1123
 
                                       const pj_sockaddr_t *addr,
1124
 
                                       int addrlen)
1125
 
{
1126
 
    int rc;
1127
 
    DWORD bytesWritten;
1128
 
    DWORD dwFlags;
1129
 
    union operation_key *op_key_rec;
1130
 
 
1131
 
    PJ_CHECK_STACK();
1132
 
    PJ_ASSERT_RETURN(key && op_key && data, PJ_EINVAL);
1133
 
 
1134
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
1135
 
    /* Check key is not closing */
1136
 
    if (key->closing)
1137
 
        return PJ_ECANCELLED;
1138
 
#endif
1139
 
 
1140
 
    op_key_rec = (union operation_key*)op_key->internal__;
1141
 
 
1142
 
    /*
1143
 
     * First try blocking write.
1144
 
     */
1145
 
    op_key_rec->overlapped.wsabuf.buf = (void*)data;
1146
 
    op_key_rec->overlapped.wsabuf.len = *length;
1147
 
 
1148
 
    dwFlags = flags;
1149
 
 
1150
 
    if ((flags & PJ_IOQUEUE_ALWAYS_ASYNC) == 0) {
1151
 
        rc = WSASendTo((SOCKET)key->hnd, &op_key_rec->overlapped.wsabuf, 1,
1152
 
                       &bytesWritten, dwFlags, addr, addrlen,
1153
 
                       NULL, NULL);
1154
 
        if (rc == 0) {
1155
 
            *length = bytesWritten;
1156
 
            return PJ_SUCCESS;
1157
 
        } else {
1158
 
            DWORD dwStatus = WSAGetLastError();
1159
 
            if (dwStatus != WSAEWOULDBLOCK) {
1160
 
                *length = -1;
1161
 
                return PJ_RETURN_OS_ERROR(dwStatus);
1162
 
            }
1163
 
        }
1164
 
    }
1165
 
 
1166
 
    dwFlags &= ~(PJ_IOQUEUE_ALWAYS_ASYNC);
1167
 
 
1168
 
    /*
1169
 
     * Data can't be sent immediately.
1170
 
     * Schedule asynchronous WSASend().
1171
 
     */
1172
 
    pj_bzero( &op_key_rec->overlapped.overlapped,
1173
 
              sizeof(op_key_rec->overlapped.overlapped));
1174
 
    op_key_rec->overlapped.operation = PJ_IOQUEUE_OP_SEND;
1175
 
 
1176
 
    rc = WSASendTo((SOCKET)key->hnd, &op_key_rec->overlapped.wsabuf, 1,
1177
 
                   &bytesWritten,  dwFlags, addr, addrlen,
1178
 
                   &op_key_rec->overlapped.overlapped, NULL);
1179
 
    if (rc == SOCKET_ERROR) {
1180
 
        DWORD dwStatus = WSAGetLastError();
1181
 
        if (dwStatus!=WSA_IO_PENDING)
1182
 
            return PJ_STATUS_FROM_OS(dwStatus);
1183
 
    }
1184
 
 
1185
 
    /* Asynchronous operation successfully submitted. */
1186
 
    return PJ_EPENDING;
1187
 
}
1188
 
 
1189
 
#if PJ_HAS_TCP
1190
 
 
1191
 
/*
1192
 
 * pj_ioqueue_accept()
1193
 
 *
1194
 
 * Initiate overlapped accept() operation.
1195
 
 */
1196
 
PJ_DEF(pj_status_t) pj_ioqueue_accept( pj_ioqueue_key_t *key,
1197
 
                                       pj_ioqueue_op_key_t *op_key,
1198
 
                                       pj_sock_t *new_sock,
1199
 
                                       pj_sockaddr_t *local,
1200
 
                                       pj_sockaddr_t *remote,
1201
 
                                       int *addrlen)
1202
 
{
1203
 
    BOOL rc;
1204
 
    DWORD bytesReceived;
1205
 
    pj_status_t status;
1206
 
    union operation_key *op_key_rec;
1207
 
    SOCKET sock;
1208
 
 
1209
 
    PJ_CHECK_STACK();
1210
 
    PJ_ASSERT_RETURN(key && op_key && new_sock, PJ_EINVAL);
1211
 
 
1212
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
1213
 
    /* Check key is not closing */
1214
 
    if (key->closing)
1215
 
        return PJ_ECANCELLED;
1216
 
#endif
1217
 
 
1218
 
    /*
1219
 
     * See if there is a new connection immediately available.
1220
 
     */
1221
 
    sock = WSAAccept((SOCKET)key->hnd, remote, addrlen, NULL, 0);
1222
 
    if (sock != INVALID_SOCKET) {
1223
 
        /* Yes! New socket is available! */
1224
 
        if (local && addrlen) {
1225
 
            int status;
1226
 
 
1227
 
            /* On WinXP or later, use SO_UPDATE_ACCEPT_CONTEXT so that socket
1228
 
             * addresses can be obtained with getsockname() and getpeername().
1229
 
             */
1230
 
            status = setsockopt(sock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
1231
 
                                (char*)&key->hnd, sizeof(SOCKET));
1232
 
            /* SO_UPDATE_ACCEPT_CONTEXT is for WinXP or later.
1233
 
             * So ignore the error status.
1234
 
             */
1235
 
 
1236
 
            status = getsockname(sock, local, addrlen);
1237
 
            if (status != 0) {
1238
 
                DWORD dwError = WSAGetLastError();
1239
 
                closesocket(sock);
1240
 
                return PJ_RETURN_OS_ERROR(dwError);
1241
 
            }
1242
 
        }
1243
 
 
1244
 
        *new_sock = sock;
1245
 
        return PJ_SUCCESS;
1246
 
 
1247
 
    } else {
1248
 
        DWORD dwError = WSAGetLastError();
1249
 
        if (dwError != WSAEWOULDBLOCK) {
1250
 
            return PJ_RETURN_OS_ERROR(dwError);
1251
 
        }
1252
 
    }
1253
 
 
1254
 
    /*
1255
 
     * No connection is immediately available.
1256
 
     * Must schedule an asynchronous operation.
1257
 
     */
1258
 
    op_key_rec = (union operation_key*)op_key->internal__;
1259
 
 
1260
 
    status = pj_sock_socket(pj_AF_INET(), pj_SOCK_STREAM(), 0,
1261
 
                            &op_key_rec->accept.newsock);
1262
 
    if (status != PJ_SUCCESS)
1263
 
        return status;
1264
 
 
1265
 
    op_key_rec->accept.operation = PJ_IOQUEUE_OP_ACCEPT;
1266
 
    op_key_rec->accept.addrlen = addrlen;
1267
 
    op_key_rec->accept.local = local;
1268
 
    op_key_rec->accept.remote = remote;
1269
 
    op_key_rec->accept.newsock_ptr = new_sock;
1270
 
    pj_bzero( &op_key_rec->accept.overlapped,
1271
 
              sizeof(op_key_rec->accept.overlapped));
1272
 
 
1273
 
    rc = AcceptEx( (SOCKET)key->hnd, (SOCKET)op_key_rec->accept.newsock,
1274
 
                   op_key_rec->accept.accept_buf,
1275
 
                   0, ACCEPT_ADDR_LEN, ACCEPT_ADDR_LEN,
1276
 
                   &bytesReceived,
1277
 
                   &op_key_rec->accept.overlapped );
1278
 
 
1279
 
    if (rc == TRUE) {
1280
 
        ioqueue_on_accept_complete(key, &op_key_rec->accept);
1281
 
        return PJ_SUCCESS;
1282
 
    } else {
1283
 
        DWORD dwStatus = WSAGetLastError();
1284
 
        if (dwStatus!=WSA_IO_PENDING)
1285
 
            return PJ_STATUS_FROM_OS(dwStatus);
1286
 
    }
1287
 
 
1288
 
    /* Asynchronous Accept() has been submitted. */
1289
 
    return PJ_EPENDING;
1290
 
}
1291
 
 
1292
 
 
1293
 
/*
1294
 
 * pj_ioqueue_connect()
1295
 
 *
1296
 
 * Initiate overlapped connect() operation (well, it's non-blocking actually,
1297
 
 * since there's no overlapped version of connect()).
1298
 
 */
1299
 
PJ_DEF(pj_status_t) pj_ioqueue_connect( pj_ioqueue_key_t *key,
1300
 
                                        const pj_sockaddr_t *addr,
1301
 
                                        int addrlen )
1302
 
{
1303
 
    HANDLE hEvent;
1304
 
    pj_ioqueue_t *ioqueue;
1305
 
 
1306
 
    PJ_CHECK_STACK();
1307
 
    PJ_ASSERT_RETURN(key && addr && addrlen, PJ_EINVAL);
1308
 
 
1309
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
1310
 
    /* Check key is not closing */
1311
 
    if (key->closing)
1312
 
        return PJ_ECANCELLED;
1313
 
#endif
1314
 
 
1315
 
    /* Initiate connect() */
1316
 
    if (connect((pj_sock_t)key->hnd, addr, addrlen) != 0) {
1317
 
        DWORD dwStatus;
1318
 
        dwStatus = WSAGetLastError();
1319
 
        if (dwStatus != WSAEWOULDBLOCK) {
1320
 
            return PJ_RETURN_OS_ERROR(dwStatus);
1321
 
        }
1322
 
    } else {
1323
 
        /* Connect has completed immediately! */
1324
 
        return PJ_SUCCESS;
1325
 
    }
1326
 
 
1327
 
    ioqueue = key->ioqueue;
1328
 
 
1329
 
    /* Add to the array of connecting socket to be polled */
1330
 
    pj_lock_acquire(ioqueue->lock);
1331
 
 
1332
 
    if (ioqueue->connecting_count >= MAXIMUM_WAIT_OBJECTS) {
1333
 
        pj_lock_release(ioqueue->lock);
1334
 
        return PJ_ETOOMANYCONN;
1335
 
    }
1336
 
 
1337
 
    /* Get or create event object. */
1338
 
    if (ioqueue->event_count) {
1339
 
        hEvent = ioqueue->event_pool[ioqueue->event_count - 1];
1340
 
        --ioqueue->event_count;
1341
 
    } else {
1342
 
        hEvent = CreateEvent(NULL, TRUE, FALSE, NULL);
1343
 
        if (hEvent == NULL) {
1344
 
            DWORD dwStatus = GetLastError();
1345
 
            pj_lock_release(ioqueue->lock);
1346
 
            return PJ_STATUS_FROM_OS(dwStatus);
1347
 
        }
1348
 
    }
1349
 
 
1350
 
    /* Mark key as connecting.
1351
 
     * We can't use array index since key can be removed dynamically.
1352
 
     */
1353
 
    key->connecting = 1;
1354
 
 
1355
 
    /* Associate socket events to the event object. */
1356
 
    if (WSAEventSelect((pj_sock_t)key->hnd, hEvent, FD_CONNECT) != 0) {
1357
 
        CloseHandle(hEvent);
1358
 
        pj_lock_release(ioqueue->lock);
1359
 
        return PJ_RETURN_OS_ERROR(WSAGetLastError());
1360
 
    }
1361
 
 
1362
 
    /* Add to array. */
1363
 
    ioqueue->connecting_keys[ ioqueue->connecting_count ] = key;
1364
 
    ioqueue->connecting_handles[ ioqueue->connecting_count ] = hEvent;
1365
 
    ioqueue->connecting_count++;
1366
 
 
1367
 
    pj_lock_release(ioqueue->lock);
1368
 
 
1369
 
    return PJ_EPENDING;
1370
 
}
1371
 
#endif  /* #if PJ_HAS_TCP */
1372
 
 
1373
 
 
1374
 
PJ_DEF(void) pj_ioqueue_op_key_init( pj_ioqueue_op_key_t *op_key,
1375
 
                                     pj_size_t size )
1376
 
{
1377
 
    pj_bzero(op_key, size);
1378
 
}
1379
 
 
1380
 
PJ_DEF(pj_bool_t) pj_ioqueue_is_pending( pj_ioqueue_key_t *key,
1381
 
                                         pj_ioqueue_op_key_t *op_key )
1382
 
{
1383
 
    BOOL rc;
1384
 
    DWORD bytesTransfered;
1385
 
 
1386
 
    rc = GetOverlappedResult( key->hnd, (LPOVERLAPPED)op_key,
1387
 
                              &bytesTransfered, FALSE );
1388
 
 
1389
 
    if (rc == FALSE) {
1390
 
        return GetLastError()==ERROR_IO_INCOMPLETE;
1391
 
    }
1392
 
 
1393
 
    return FALSE;
1394
 
}
1395
 
 
1396
 
 
1397
 
PJ_DEF(pj_status_t) pj_ioqueue_post_completion( pj_ioqueue_key_t *key,
1398
 
                                                pj_ioqueue_op_key_t *op_key,
1399
 
                                                pj_ssize_t bytes_status )
1400
 
{
1401
 
    BOOL rc;
1402
 
 
1403
 
    rc = PostQueuedCompletionStatus(key->ioqueue->iocp, bytes_status,
1404
 
                                    (long)key, (OVERLAPPED*)op_key );
1405
 
    if (rc == FALSE) {
1406
 
        return PJ_RETURN_OS_ERROR(GetLastError());
1407
 
    }
1408
 
 
1409
 
    return PJ_SUCCESS;
1410
 
}
1411
 
 
1412
 
PJ_DEF(pj_status_t) pj_ioqueue_set_concurrency(pj_ioqueue_key_t *key,
1413
 
                                               pj_bool_t allow)
1414
 
{
1415
 
    PJ_ASSERT_RETURN(key, PJ_EINVAL);
1416
 
 
1417
 
    /* PJ_IOQUEUE_HAS_SAFE_UNREG must be enabled if concurrency is
1418
 
     * disabled.
1419
 
     */
1420
 
    PJ_ASSERT_RETURN(allow || PJ_IOQUEUE_HAS_SAFE_UNREG, PJ_EINVAL);
1421
 
 
1422
 
    key->allow_concurrent = allow;
1423
 
    return PJ_SUCCESS;
1424
 
}
1425
 
 
1426
 
PJ_DEF(pj_status_t) pj_ioqueue_lock_key(pj_ioqueue_key_t *key)
1427
 
{
1428
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
1429
 
    return pj_mutex_lock(key->mutex);
1430
 
#else
1431
 
    PJ_ASSERT_RETURN(!"PJ_IOQUEUE_HAS_SAFE_UNREG is disabled", PJ_EINVALIDOP);
1432
 
#endif
1433
 
}
1434
 
 
1435
 
PJ_DEF(pj_status_t) pj_ioqueue_unlock_key(pj_ioqueue_key_t *key)
1436
 
{
1437
 
#if PJ_IOQUEUE_HAS_SAFE_UNREG
1438
 
    return pj_mutex_unlock(key->mutex);
1439
 
#else
1440
 
    PJ_ASSERT_RETURN(!"PJ_IOQUEUE_HAS_SAFE_UNREG is disabled", PJ_EINVALIDOP);
1441
 
#endif
1442
 
}