Subversion Repositories HelenOS

Rev

Rev 1408 | Rev 1435 | Go to most recent revision | Show entire file | Ignore whitespace | Details | Blame | Last modification | View Log | RSS feed

Rev 1408 Rev 1427
Line 89... Line 89...
89
#include <libadt/list.h>
89
#include <libadt/list.h>
90
#include <ipc/ipc.h>
90
#include <ipc/ipc.h>
91
#include <assert.h>
91
#include <assert.h>
92
#include <errno.h>
92
#include <errno.h>
93
 
93
 
94
static atomic_t conn_futex = FUTEX_INITIALIZER;
94
static atomic_t async_futex = FUTEX_INITIALIZER;
95
static hash_table_t conn_hash_table;
95
static hash_table_t conn_hash_table;
96
 
96
 
97
typedef struct {
97
typedef struct {
-
 
98
    pstid_t ptid;                /**< Thread waiting for this message */
-
 
99
    int active;                  /**< If this thread is currently active */
-
 
100
    int done;                    /**< If reply was received */
-
 
101
    ipc_call_t *dataptr;         /**< Pointer where the answer data
-
 
102
                      *   should be stored */
-
 
103
    ipcarg_t retval;
-
 
104
} amsg_t;
-
 
105
 
-
 
106
typedef struct {
98
    link_t link;
107
    link_t link;
99
    ipc_callid_t callid;
108
    ipc_callid_t callid;
100
    ipc_call_t call;
109
    ipc_call_t call;
101
} msg_t;
110
} msg_t;
102
 
111
 
Line 113... Line 122...
113
} connection_t;
122
} connection_t;
114
 
123
 
115
__thread connection_t *PS_connection;
124
__thread connection_t *PS_connection;
116
 
125
 
117
/* Hash table functions */
126
/* Hash table functions */
118
 
-
 
119
#define CONN_HASH_TABLE_CHAINS  32
127
#define CONN_HASH_TABLE_CHAINS  32
120
 
128
 
121
static hash_index_t conn_hash(unsigned long *key)
129
static hash_index_t conn_hash(unsigned long *key)
122
{
130
{
123
    assert(key);
131
    assert(key);
Line 144... Line 152...
144
    .hash = conn_hash,
152
    .hash = conn_hash,
145
    .compare = conn_compare,
153
    .compare = conn_compare,
146
    .remove_callback = conn_remove
154
    .remove_callback = conn_remove
147
};
155
};
148
 
156
 
-
 
157
/*************************************************/
-
 
158
 
149
/** Try to route a call to an appropriate connection thread
159
/** Try to route a call to an appropriate connection thread
150
 *
160
 *
151
 */
161
 */
152
static int route_call(ipc_callid_t callid, ipc_call_t *call)
162
static int route_call(ipc_callid_t callid, ipc_call_t *call)
153
{
163
{
154
    connection_t *conn;
164
    connection_t *conn;
155
    msg_t *msg;
165
    msg_t *msg;
156
    link_t *hlp;
166
    link_t *hlp;
157
    unsigned long key;
167
    unsigned long key;
158
 
168
 
159
    futex_down(&conn_futex);
169
    futex_down(&async_futex);
160
 
170
 
161
    key = call->in_phone_hash;
171
    key = call->in_phone_hash;
162
    hlp = hash_table_find(&conn_hash_table, &key);
172
    hlp = hash_table_find(&conn_hash_table, &key);
163
    if (!hlp) {
173
    if (!hlp) {
164
        futex_up(&conn_futex);
174
        futex_up(&async_futex);
165
        return 0;
175
        return 0;
166
    }
176
    }
167
    conn = hash_table_get_instance(hlp, connection_t, link);
177
    conn = hash_table_get_instance(hlp, connection_t, link);
168
 
178
 
169
    msg = malloc(sizeof(*msg));
179
    msg = malloc(sizeof(*msg));
Line 174... Line 184...
174
    if (!conn->active) {
184
    if (!conn->active) {
175
        conn->active = 1;
185
        conn->active = 1;
176
        psthread_add_ready(conn->ptid);
186
        psthread_add_ready(conn->ptid);
177
    }
187
    }
178
 
188
 
179
    futex_up(&conn_futex);
189
    futex_up(&async_futex);
180
 
190
 
181
    return 1;
191
    return 1;
182
}
192
}
183
 
193
 
184
/** Return new incoming message for current(thread-local) connection */
194
/** Return new incoming message for current(thread-local) connection */
Line 186... Line 196...
186
{
196
{
187
    msg_t *msg;
197
    msg_t *msg;
188
    ipc_callid_t callid;
198
    ipc_callid_t callid;
189
    connection_t *conn;
199
    connection_t *conn;
190
   
200
   
191
    futex_down(&conn_futex);
201
    futex_down(&async_futex);
192
 
202
 
193
    conn = PS_connection;
203
    conn = PS_connection;
194
    /* If nothing in queue, wait until something appears */
204
    /* If nothing in queue, wait until something appears */
195
    if (list_empty(&conn->msg_queue)) {
205
    if (list_empty(&conn->msg_queue)) {
196
        conn->active = 0;
206
        conn->active = 0;
Line 201... Line 211...
201
    list_remove(&msg->link);
211
    list_remove(&msg->link);
202
    callid = msg->callid;
212
    callid = msg->callid;
203
    *call = msg->call;
213
    *call = msg->call;
204
    free(msg);
214
    free(msg);
205
   
215
   
206
    futex_up(&conn_futex);
216
    futex_up(&async_futex);
207
    return callid;
217
    return callid;
208
}
218
}
209
 
219
 
210
/** Thread function that gets created on new connection
220
/** Thread function that gets created on new connection
211
 *
221
 *
Line 234... Line 244...
234
    PS_connection = (connection_t *)arg;
244
    PS_connection = (connection_t *)arg;
235
    conn = PS_connection;
245
    conn = PS_connection;
236
    conn->cthread(conn->callid, &conn->call);
246
    conn->cthread(conn->callid, &conn->call);
237
 
247
 
238
    /* Remove myself from connection hash table */
248
    /* Remove myself from connection hash table */
239
    futex_down(&conn_futex);
249
    futex_down(&async_futex);
240
    key = conn->in_phone_hash;
250
    key = conn->in_phone_hash;
241
    hash_table_remove(&conn_hash_table, &key, 1);
251
    hash_table_remove(&conn_hash_table, &key, 1);
242
    futex_up(&conn_futex);
252
    futex_up(&async_futex);
243
    /* Answer all remaining messages with ehangup */
253
    /* Answer all remaining messages with ehangup */
244
    while (!list_empty(&conn->msg_queue)) {
254
    while (!list_empty(&conn->msg_queue)) {
245
        msg = list_get_instance(conn->msg_queue.next, msg_t, link);
255
        msg = list_get_instance(conn->msg_queue.next, msg_t, link);
246
        list_remove(&msg->link);
256
        list_remove(&msg->link);
247
        ipc_answer_fast(msg->callid, EHANGUP, 0, 0);
257
        ipc_answer_fast(msg->callid, EHANGUP, 0, 0);
Line 286... Line 296...
286
        free(conn);
296
        free(conn);
287
        ipc_answer_fast(callid, ENOMEM, 0, 0);
297
        ipc_answer_fast(callid, ENOMEM, 0, 0);
288
        return NULL;
298
        return NULL;
289
    }
299
    }
290
    key = conn->in_phone_hash;
300
    key = conn->in_phone_hash;
291
    futex_down(&conn_futex);
301
    futex_down(&async_futex);
292
    /* Add connection to hash table */
302
    /* Add connection to hash table */
293
    hash_table_insert(&conn_hash_table, &key, &conn->link);
303
    hash_table_insert(&conn_hash_table, &key, &conn->link);
294
    futex_up(&conn_futex);
304
    futex_up(&async_futex);
295
 
305
 
296
    psthread_add_ready(conn->ptid);
306
    psthread_add_ready(conn->ptid);
297
 
307
 
298
    return conn->ptid;
308
    return conn->ptid;
299
}
309
}
300
 
310
 
301
/** Handle call to a task */
311
/** Handle call that was received */
302
static void handle_call(ipc_callid_t callid, ipc_call_t *call)
312
static void handle_call(ipc_callid_t callid, ipc_call_t *call)
303
{
313
{
304
    if (route_call(callid, call))
314
    if (route_call(callid, call))
305
        return;
315
        return;
306
 
316
 
Line 322... Line 332...
322
    ipc_call_t call;
332
    ipc_call_t call;
323
    ipc_callid_t callid;
333
    ipc_callid_t callid;
324
 
334
 
325
    while (1) {
335
    while (1) {
326
        if (psthread_schedule_next_adv(PS_FROM_MANAGER)) {
336
        if (psthread_schedule_next_adv(PS_FROM_MANAGER)) {
327
            futex_up(&conn_futex); /* conn_futex is always held
337
            futex_up(&async_futex); /* async_futex is always held
328
                        * when entering manager thread
338
                        * when entering manager thread
329
                        */
339
                        */
330
            continue;
340
            continue;
331
        }
341
        }
332
        callid = ipc_wait_cycle(&call,SYNCH_NO_TIMEOUT,SYNCH_BLOCKING);
342
        callid = ipc_wait_cycle(&call,SYNCH_NO_TIMEOUT,SYNCH_BLOCKING);
333
 
343
 
334
        if (callid & IPC_CALLID_ANSWERED)
344
        if (callid & IPC_CALLID_ANSWERED)
335
            continue;
345
            continue;
-
 
346
 
336
        handle_call(callid, &call);
347
        handle_call(callid, &call);
337
    }
348
    }
338
}
349
}
339
 
350
 
340
/** Function to start async_manager as a standalone thread
351
/** Function to start async_manager as a standalone thread
Line 344... Line 355...
344
 * currently one async_manager is started automatically per kernel
355
 * currently one async_manager is started automatically per kernel
345
 * thread except main thread.
356
 * thread except main thread.
346
 */
357
 */
347
static int async_manager_thread(void *arg)
358
static int async_manager_thread(void *arg)
348
{
359
{
349
    futex_up(&conn_futex); /* conn_futex is always locked when entering
360
    futex_up(&async_futex); /* async_futex is always locked when entering
350
                * manager */
361
                * manager */
351
    async_manager();
362
    async_manager();
352
}
363
}
353
 
364
 
354
/** Add one manager to manager list */
365
/** Add one manager to manager list */
Line 373... Line 384...
373
        printf("%s: cannot create hash table\n", "async");
384
        printf("%s: cannot create hash table\n", "async");
374
        return ENOMEM;
385
        return ENOMEM;
375
    }
386
    }
376
   
387
   
377
}
388
}
-
 
389
 
-
 
390
/** IPC handler for messages in async framework
-
 
391
 *
-
 
392
 * Notify thread that is waiting for this message, that it arrived
-
 
393
 */
-
 
394
static void reply_received(void *private, int retval,
-
 
395
               ipc_call_t *data)
-
 
396
{
-
 
397
    amsg_t *msg = (amsg_t *) private;
-
 
398
 
-
 
399
    msg->retval = retval;
-
 
400
 
-
 
401
    futex_down(&async_futex);
-
 
402
    /* Copy data after futex_down, just in case the
-
 
403
     * call was detached
-
 
404
     */
-
 
405
    if (msg->dataptr)
-
 
406
        *msg->dataptr = *data;
-
 
407
   
-
 
408
    msg->done = 1;
-
 
409
    if (! msg->active) {
-
 
410
        msg->active = 1;
-
 
411
        psthread_add_ready(msg->ptid);
-
 
412
    }
-
 
413
    futex_up(&async_futex);
-
 
414
}
-
 
415
 
-
 
416
/** Send message and return id of the sent message
-
 
417
 *
-
 
418
 * The return value can be used as input for async_wait() to wait
-
 
419
 * for completion.
-
 
420
 */
-
 
421
aid_t async_send_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2,
-
 
422
           ipc_call_t *dataptr)
-
 
423
{
-
 
424
    amsg_t *msg;
-
 
425
 
-
 
426
    msg = malloc(sizeof(*msg));
-
 
427
    msg->active = 1;
-
 
428
    msg->done = 0;
-
 
429
    msg->dataptr = dataptr;
-
 
430
    ipc_call_async_2(phoneid,method,arg1,arg2,msg,reply_received);
-
 
431
 
-
 
432
    return (aid_t) msg;
-
 
433
}
-
 
434
 
-
 
435
/** Wait for a message sent by async framework
-
 
436
 *
-
 
437
 * @param amsgid Message ID to wait for
-
 
438
 * @param retval Pointer to variable where will be stored retval
-
 
439
 *               of the answered message. If NULL, it is ignored.
-
 
440
 *
-
 
441
 */
-
 
442
void async_wait_for(aid_t amsgid, ipcarg_t *retval)
-
 
443
{
-
 
444
    amsg_t *msg = (amsg_t *) amsgid;
-
 
445
    connection_t *conn;
-
 
446
 
-
 
447
    futex_down(&async_futex);
-
 
448
    if (msg->done) {
-
 
449
        futex_up(&async_futex);
-
 
450
        goto done;
-
 
451
    }
-
 
452
 
-
 
453
    msg->ptid = psthread_get_id();
-
 
454
    msg->active = 0;
-
 
455
    /* Leave locked async_futex when entering this function */
-
 
456
    psthread_schedule_next_adv(PS_TO_MANAGER);
-
 
457
    /* futex is up automatically after psthread_schedule_next...*/
-
 
458
done:
-
 
459
    if (retval)
-
 
460
        *retval = msg->retval;
-
 
461
    free(msg);
-
 
462
}