WebSVN – HelenOS – Diff – /trunk/uspace/lib/libc/generic/async.c



 *
 *       callid = async_get_call(&call);
 *       ....
 * }
 *
 
 */
#include <futex.h>
#include <async.h>
#include <fibril.h>
#include <stdio.h>
#include <libadt/hash_table.h>
#include <libadt/list.h>
#include <ipc/ipc.h>
#include <assert.h>

atomic_t async_futex = FUTEX_INITIALIZER;
static hash_table_t conn_hash_table;
static LIST_INITIALIZE(timeout_list);
 
typedef struct {
    /** Expiration time for waiting fibril. */
    struct timeval expires;    
    /** If true, this struct is in the timeout list. */
    int inlist;
    link_t link;
 
    /** Fibril waiting for this message. */
    fid_t fid;
    /** If this fibril is currently active. */
    int active;
    /** If true, we timed out. */
    int timedout;
} awaiter_t;
 
typedef struct {
    awaiter_t wdata;
 
    int done;                   /**< If reply was received */
    ipc_call_t *dataptr;        /**< Pointer where the answer data
                     *   is stored */
    ipcarg_t retval;
} amsg_t;
 
typedef struct {
    link_t link;

} msg_t;
 
typedef struct {
    awaiter_t wdata;
 
    link_t link;            /**< Hash table link. */
    ipcarg_t in_phone_hash;     /**< Incoming phone hash. */
    link_t msg_queue;       /**< Messages that should be delivered
                     *   to this fibril. */
    /* Structures for connection opening packet */
    ipc_callid_t callid;
    ipc_call_t call;
    ipc_callid_t close_callid;  /* Identification of closing packet. */
    void (*cfibril)(ipc_callid_t, ipc_call_t *);
} connection_t;
 
/** Identifier of the incoming connection handled by the current fibril. */
__thread connection_t *FIBRIL_connection;
/** If true, it is forbidden to use async_req functions and
 *  all preemption is disabled */
__thread int in_interrupt_handler;
 
static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);

        if (conn->wdata.inlist) {
            conn->wdata.inlist = 0;
            list_remove(&conn->wdata.link);
        }
        conn->wdata.active = 1;
        fibril_add_ready(conn->wdata.fid);
    }
 
    futex_up(&async_futex);
 
    return 1;

{
    msg_t *msg;
    ipc_callid_t callid;
    connection_t *conn;
   
    assert(FIBRIL_connection);
    /* GCC 4.1.0 coughs on FIBRIL_connection-> dereference,
     * GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
     *           I would never expect to find so many errors in
     *           compiler *($&$(*&$
     */
    conn = FIBRIL_connection;
 
    futex_down(&async_futex);
 
    if (usecs) {
        gettimeofday(&conn->wdata.expires, NULL);

    while (list_empty(&conn->msg_queue)) {
        if (usecs)
            insert_timeout(&conn->wdata);
 
        conn->wdata.active = 0;
        fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
        /* Futex is up after getting back from async_manager
         * get it again */
        futex_down(&async_futex);
        if (usecs && conn->wdata.timedout &&
            list_empty(&conn->msg_queue)) {
            /* If we timed out-> exit */
            futex_up(&async_futex);
            return 0;
        }

 * When new connection arrives, thread with this function is created.
 * It calls client_connection and does final cleanup.
 *
 * @param arg Connection structure pointer
 */
static int connection_fibril(void  *arg)
{
    unsigned long key;
    msg_t *msg;
    int close_answered = 0;
 
    /* Setup thread local connection pointer */
    FIBRIL_connection = (connection_t *) arg;
    FIBRIL_connection->cfibril(FIBRIL_connection->callid,
        &FIBRIL_connection->call);
   
    /* Remove myself from connection hash table */
    futex_down(&async_futex);
    key = FIBRIL_connection->in_phone_hash;
    hash_table_remove(&conn_hash_table, &key, 1);
    futex_up(&async_futex);
   
    /* Answer all remaining messages with ehangup */
    while (!list_empty(&FIBRIL_connection->msg_queue)) {
        msg = list_get_instance(FIBRIL_connection->msg_queue.next,
            msg_t, link);
        list_remove(&msg->link);
        if (msg->callid == FIBRIL_connection->close_callid)
            close_answered = 1;
        ipc_answer_fast(msg->callid, EHANGUP, 0, 0);
        free(msg);
    }
    if (FIBRIL_connection->close_callid)
        ipc_answer_fast(FIBRIL_connection->close_callid, 0, 0, 0);
   
    return 0;
}
 
/** Create new thread for a new connection

 * threads.
 *
 * @param in_phone_hash Identification of the incoming connection
 * @param callid Callid of the IPC_M_CONNECT_ME_TO packet
 * @param call Call data of the opening packet
 * @param cfibril Fibril function that should be called upon
 *                opening the connection
 * @return New fibril id.
 */
fid_t async_new_connection(ipcarg_t in_phone_hash, ipc_callid_t callid,
    ipc_call_t *call, void (*cfibril)(ipc_callid_t, ipc_call_t *))
{
    connection_t *conn;
    unsigned long key;
 
    conn = malloc(sizeof(*conn));

    conn->callid = callid;
    conn->close_callid = 0;
    if (call)
        conn->call = *call;
    conn->wdata.active = 1; /* We will activate it asap */
    conn->cfibril = cfibril;
 
    conn->wdata.fid = fibril_create(connection_fibril, conn);
    if (!conn->wdata.fid) {
        free(conn);
        ipc_answer_fast(callid, ENOMEM, 0, 0);
        return NULL;
    }
    /* Add connection to hash table */
    key = conn->in_phone_hash;
    futex_down(&async_futex);
    hash_table_insert(&conn_hash_table, &key, &conn->link);
    futex_up(&async_futex);
 
    fibril_add_ready(conn->wdata.fid);
 
    return conn->wdata.fid;
}
 
/** Handle call that was received */
static void handle_call(ipc_callid_t callid, ipc_call_t *call)
{

    }      
 
    switch (IPC_GET_METHOD(*call)) {
    case IPC_M_CONNECT_ME_TO:
        /* Open new connection with thread etc. */
        async_new_connection(IPC_GET_ARG3(*call), callid, call,
            client_connection);
        return;
    }
 
    /* Try to route call through connection tables */
    if (route_call(callid, call))

    gettimeofday(&tv,NULL);
    futex_down(&async_futex);
 
    cur = timeout_list.next;
    while (cur != &timeout_list) {
        waiter = list_get_instance(cur, awaiter_t, link);
        if (tv_gt(&waiter->expires, &tv))
            break;
        cur = cur->next;
        list_remove(&waiter->link);
        waiter->inlist = 0;

        /* Redundant condition? The thread should not
         * be active when it gets here.
         */
        if (!waiter->active) {
            waiter->active = 1;
            fibril_add_ready(waiter->fid);
        }
    }
 
    futex_up(&async_futex);
}

    int timeout;
    awaiter_t *waiter;
    struct timeval tv;
 
    while (1) {
        if (fibril_schedule_next_adv(FIBRIL_FROM_MANAGER)) {
            futex_up(&async_futex);
            /* async_futex is always held
             * when entering manager thread
             */
            continue;
        }
        futex_down(&async_futex);
        if (!list_empty(&timeout_list)) {
            waiter = list_get_instance(timeout_list.next, awaiter_t,
                link);
            gettimeofday(&tv, NULL);
            if (tv_gteq(&tv, &waiter->expires)) {
                futex_up(&async_futex);
                handle_expired_timeouts();
                continue;
            } else

}
 
/** Add one manager to manager list */
void async_create_manager(void)
{
    fid_t fid;
 
    fid = fibril_create(async_manager_thread, NULL);
    fibril_add_manager(fid);
}
 
/** Remove one manager from manager list */
void async_destroy_manager(void)
{
    fibril_remove_manager();
}
 
/** Initialize internal structures needed for async manager */
int _async_init(void)
{
    if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,
        &conn_hash_table_ops)) {
        printf("%s: cannot create hash table\n", "async");
        return ENOMEM;
    }
   
    return 0;
}
 
/** IPC handler for messages in async framework
 *
 * Notify the fibril which is waiting for this message, that it arrived
 */
static void reply_received(void *private, int retval,
               ipc_call_t *data)
{
    amsg_t *msg = (amsg_t *) private;

    if (msg->wdata.inlist)
        list_remove(&msg->wdata.link);
    msg->done = 1;
    if (! msg->wdata.active) {
        msg->wdata.active = 1;
        fibril_add_ready(msg->wdata.fid);
    }
    futex_up(&async_futex);
}
 
/** Send message and return id of the sent message

           ipc_call_t *dataptr)
{
    amsg_t *msg;
 
    if (in_interrupt_handler) {
        printf("Cannot send asynchronous request in interrupt "
            "handler.\n");
        _exit(1);
    }
 
    msg = malloc(sizeof(*msg));
    msg->done = 0;
    msg->dataptr = dataptr;
 
    msg->wdata.active = 1; /* We may sleep in next method, but it
                * will use it's own mechanism */
    ipc_call_async_2(phoneid, method, arg1, arg2, msg, reply_received, 1);
 
    return (aid_t) msg;
}
 
/** Send message and return id of the sent message

    msg->done = 0;
    msg->dataptr = dataptr;
 
    msg->wdata.active = 1; /* We may sleep in next method, but it
                * will use it's own mechanism */
    ipc_call_async_3(phoneid, method, arg1, arg2, arg3, msg, reply_received,
        1);
 
    return (aid_t) msg;
}
 
/** Wait for a message sent by async framework

    if (msg->done) {
        futex_up(&async_futex);
        goto done;
    }
 
    msg->wdata.fid = fibril_get_id();
    msg->wdata.active = 0;
    msg->wdata.inlist = 0;
    /* Leave locked async_futex when entering this function */
    fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
    /* futex is up automatically after fibril_schedule_next...*/
done:
    if (retval)
        *retval = msg->retval;
    free(msg);
}

    }
 
    gettimeofday(&msg->wdata.expires, NULL);
    tv_add(&msg->wdata.expires, timeout);
 
    msg->wdata.fid = fibril_get_id();
    msg->wdata.active = 0;
    insert_timeout(&msg->wdata);
 
    /* Leave locked async_futex when entering this function */
    fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
    /* futex is up automatically after fibril_schedule_next...*/
 
    if (!msg->done)
        return ETIMEOUT;
 
done:

 
    msg = malloc(sizeof(*msg));
    if (!msg)
        return;
 
    msg->wdata.fid = fibril_get_id();
    msg->wdata.active = 0;
 
    gettimeofday(&msg->wdata.expires, NULL);
    tv_add(&msg->wdata.expires, timeout);
 
    futex_down(&async_futex);
    insert_timeout(&msg->wdata);
    /* Leave locked async_futex when entering this function */
    fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
    /* futex is up automatically after fibril_schedule_next...*/
    free(msg);
}
 
/** Set function that is called, IPC_M_CONNECT_ME_TO is received
 *

 
/* Primitive functions for simple communication */
void async_msg_3(int phoneid, ipcarg_t method, ipcarg_t arg1,
         ipcarg_t arg2, ipcarg_t arg3)
{
    ipc_call_async_3(phoneid, method, arg1, arg2, arg3, NULL, NULL,
        !in_interrupt_handler);
}
 
void async_msg_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2)
{
    ipc_call_async_2(phoneid, method, arg1, arg2, NULL, NULL,
        !in_interrupt_handler);
}
 
/** @}
 */

Rev 2479	Rev 2482
Line 83...	Line 83...
83	*	83	*
84	* callid = async_get_call(&call);	84	* callid = async_get_call(&call);
85	* ....	85	* ....
86	* }	86	* }
87	*	87	*
88	* TODO: Detaching/joining dead psthreads?	-
89	*/	88	*/
90	#include <futex.h>	89	#include <futex.h>
91	#include <async.h>	90	#include <async.h>
92	#include <psthread.h>	91	#include <fibril.h>
93	#include <stdio.h>	92	#include <stdio.h>
94	#include <libadt/hash_table.h>	93	#include <libadt/hash_table.h>
95	#include <libadt/list.h>	94	#include <libadt/list.h>
96	#include <ipc/ipc.h>	95	#include <ipc/ipc.h>
97	#include <assert.h>	96	#include <assert.h>
Line 102...	Line 101...
102	atomic_t async_futex = FUTEX_INITIALIZER;	101	atomic_t async_futex = FUTEX_INITIALIZER;
103	static hash_table_t conn_hash_table;	102	static hash_table_t conn_hash_table;
104	static LIST_INITIALIZE(timeout_list);	103	static LIST_INITIALIZE(timeout_list);
105		104
106	typedef struct {	105	typedef struct {
107	struct timeval expires; /*< Expiration time for waiting thread /	106	/** Expiration time for waiting fibril. */
-		107	struct timeval expires;
108	int inlist; /*< If true, this struct is in timeout list /	108	/** If true, this struct is in the timeout list. */
-		109	int inlist;
109	link_t link;	110	link_t link;
110		111
111	pstid_t ptid; /*< Thread waiting for this message /	112	/** Fibril waiting for this message. */
-		113	fid_t fid;
112	int active; /*< If this thread is currently active /	114	/** If this fibril is currently active. */
-		115	int active;
113	int timedout; /*< If true, we timed out /	116	/** If true, we timed out. */
-		117	int timedout;
114	} awaiter_t;	118	} awaiter_t;
115		119
116	typedef struct {	120	typedef struct {
117	awaiter_t wdata;	121	awaiter_t wdata;
118		122
119	int done; /*< If reply was received /	123	int done; /*< If reply was received /
120	ipc_call_t dataptr; /*< Pointer where the answer data	124	ipc_call_t dataptr; /*< Pointer where the answer data
121	* is stored */	125	* is stored */
122	ipcarg_t retval;	126	ipcarg_t retval;
123	} amsg_t;	127	} amsg_t;
124		128
125	typedef struct {	129	typedef struct {
126	link_t link;	130	link_t link;
Line 129...	Line 133...
129	} msg_t;	133	} msg_t;
130		134
131	typedef struct {	135	typedef struct {
132	awaiter_t wdata;	136	awaiter_t wdata;
133		137
134	link_t link; /*< Hash table link /	138	link_t link; /*< Hash table link. /
135	ipcarg_t in_phone_hash; /*< Incoming phone hash. /	139	ipcarg_t in_phone_hash; /*< Incoming phone hash. /
136	link_t msg_queue; /*< Messages that should be delivered to this thread /	140	link_t msg_queue; /**< Messages that should be delivered
-		141	* to this fibril. */
137	/* Structures for connection opening packet */	142	/* Structures for connection opening packet */
138	ipc_callid_t callid;	143	ipc_callid_t callid;
139	ipc_call_t call;	144	ipc_call_t call;
140	ipc_callid_t close_callid; /* Identification of closing packet */	145	ipc_callid_t close_callid; /* Identification of closing packet. */
141	void (cthread)(ipc_callid_t,ipc_call_t );	146	void (cfibril)(ipc_callid_t, ipc_call_t );
142	} connection_t;	147	} connection_t;
143		148
144	/** Identifier of incoming connection handled by current thread */	149	/** Identifier of the incoming connection handled by the current fibril. */
145	__thread connection_t *PS_connection;	150	__thread connection_t *FIBRIL_connection;
146	/** If true, it is forbidden to use async_req functions and	151	/** If true, it is forbidden to use async_req functions and
147	* all preemption is disabled */	152	* all preemption is disabled */
148	__thread int in_interrupt_handler;	153	__thread int in_interrupt_handler;
149		154
150	static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);	155	static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);
Line 283...	Line 288...
283	if (conn->wdata.inlist) {	288	if (conn->wdata.inlist) {
284	conn->wdata.inlist = 0;	289	conn->wdata.inlist = 0;
285	list_remove(&conn->wdata.link);	290	list_remove(&conn->wdata.link);
286	}	291	}
287	conn->wdata.active = 1;	292	conn->wdata.active = 1;
288	psthread_add_ready(conn->wdata.ptid);	293	fibril_add_ready(conn->wdata.fid);
289	}	294	}
290		295
291	futex_up(&async_futex);	296	futex_up(&async_futex);
292		297
293	return 1;	298	return 1;
Line 298...	Line 303...
298	{	303	{
299	msg_t *msg;	304	msg_t *msg;
300	ipc_callid_t callid;	305	ipc_callid_t callid;
301	connection_t *conn;	306	connection_t *conn;
302		307
303	assert(PS_connection);	308	assert(FIBRIL_connection);
304	/* GCC 4.1.0 coughs on PS_connection-> dereference,	309	/* GCC 4.1.0 coughs on FIBRIL_connection-> dereference,
305	* GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.	310	* GCC 4.1.1 happilly puts the rdhwr instruction in delay slot.
306	* I would never expect to find so many errors in	311	* I would never expect to find so many errors in
307	* compiler ($&$(&$	312	* compiler ($&$(&$
308	*/	313	*/
309	conn = PS_connection;	314	conn = FIBRIL_connection;
310		315
311	futex_down(&async_futex);	316	futex_down(&async_futex);
312		317
313	if (usecs) {	318	if (usecs) {
314	gettimeofday(&conn->wdata.expires, NULL);	319	gettimeofday(&conn->wdata.expires, NULL);
Line 320...	Line 325...
320	while (list_empty(&conn->msg_queue)) {	325	while (list_empty(&conn->msg_queue)) {
321	if (usecs)	326	if (usecs)
322	insert_timeout(&conn->wdata);	327	insert_timeout(&conn->wdata);
323		328
324	conn->wdata.active = 0;	329	conn->wdata.active = 0;
325	psthread_schedule_next_adv(PS_TO_MANAGER);	330	fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
326	/* Futex is up after getting back from async_manager	331	/* Futex is up after getting back from async_manager
327	* get it again */	332	* get it again */
328	futex_down(&async_futex);	333	futex_down(&async_futex);
329	if (usecs && conn->wdata.timedout && \	334	if (usecs && conn->wdata.timedout &&
330	list_empty(&conn->msg_queue)) {	335	list_empty(&conn->msg_queue)) {
331	/* If we timed out-> exit */	336	/* If we timed out-> exit */
332	futex_up(&async_futex);	337	futex_up(&async_futex);
333	return 0;	338	return 0;
334	}	339	}
Line 362...	Line 367...
362	* When new connection arrives, thread with this function is created.	367	* When new connection arrives, thread with this function is created.
363	* It calls client_connection and does final cleanup.	368	* It calls client_connection and does final cleanup.
364	*	369	*
365	* @param arg Connection structure pointer	370	* @param arg Connection structure pointer
366	*/	371	*/
367	static int connection_thread(void *arg)	372	static int connection_fibril(void *arg)
368	{	373	{
369	unsigned long key;	374	unsigned long key;
370	msg_t *msg;	375	msg_t *msg;
371	int close_answered = 0;	376	int close_answered = 0;
372		377
373	/* Setup thread local connection pointer */	378	/* Setup thread local connection pointer */
374	PS_connection = (connection_t *)arg;	379	FIBRIL_connection = (connection_t *) arg;
375	PS_connection->cthread(PS_connection->callid, &PS_connection->call);	380	FIBRIL_connection->cfibril(FIBRIL_connection->callid,
-		381	&FIBRIL_connection->call);
376		382
377	/* Remove myself from connection hash table */	383	/* Remove myself from connection hash table */
378	futex_down(&async_futex);	384	futex_down(&async_futex);
379	key = PS_connection->in_phone_hash;	385	key = FIBRIL_connection->in_phone_hash;
380	hash_table_remove(&conn_hash_table, &key, 1);	386	hash_table_remove(&conn_hash_table, &key, 1);
381	futex_up(&async_futex);	387	futex_up(&async_futex);
382		388
383	/* Answer all remaining messages with ehangup */	389	/* Answer all remaining messages with ehangup */
384	while (!list_empty(&PS_connection->msg_queue)) {	390	while (!list_empty(&FIBRIL_connection->msg_queue)) {
385	msg = list_get_instance(PS_connection->msg_queue.next, msg_t, link);	391	msg = list_get_instance(FIBRIL_connection->msg_queue.next,
-		392	msg_t, link);
386	list_remove(&msg->link);	393	list_remove(&msg->link);
387	if (msg->callid == PS_connection->close_callid)	394	if (msg->callid == FIBRIL_connection->close_callid)
388	close_answered = 1;	395	close_answered = 1;
389	ipc_answer_fast(msg->callid, EHANGUP, 0, 0);	396	ipc_answer_fast(msg->callid, EHANGUP, 0, 0);
390	free(msg);	397	free(msg);
391	}	398	}
392	if (PS_connection->close_callid)	399	if (FIBRIL_connection->close_callid)
393	ipc_answer_fast(PS_connection->close_callid, 0, 0, 0);	400	ipc_answer_fast(FIBRIL_connection->close_callid, 0, 0, 0);
394		401
395	return 0;	402	return 0;
396	}	403	}
397		404
398	/** Create new thread for a new connection	405	/** Create new thread for a new connection
Line 403...	Line 410...
403	* threads.	410	* threads.
404	*	411	*
405	* @param in_phone_hash Identification of the incoming connection	412	* @param in_phone_hash Identification of the incoming connection
406	* @param callid Callid of the IPC_M_CONNECT_ME_TO packet	413	* @param callid Callid of the IPC_M_CONNECT_ME_TO packet
407	* @param call Call data of the opening packet	414	* @param call Call data of the opening packet
408	* @param cthread Thread function that should be called upon	415	* @param cfibril Fibril function that should be called upon
409	* opening the connection	416	* opening the connection
410	* @return New thread id	417	* @return New fibril id.
411	*/	418	*/
412	pstid_t async_new_connection(ipcarg_t in_phone_hash,ipc_callid_t callid, ipc_call_t call, void (cthread)(ipc_callid_t, ipc_call_t *))	419	fid_t async_new_connection(ipcarg_t in_phone_hash, ipc_callid_t callid,
-		420	ipc_call_t call, void (cfibril)(ipc_callid_t, ipc_call_t *))
413	{	421	{
414	connection_t *conn;	422	connection_t *conn;
415	unsigned long key;	423	unsigned long key;
416		424
417	conn = malloc(sizeof(*conn));	425	conn = malloc(sizeof(*conn));
Line 424...	Line 432...
424	conn->callid = callid;	432	conn->callid = callid;
425	conn->close_callid = 0;	433	conn->close_callid = 0;
426	if (call)	434	if (call)
427	conn->call = *call;	435	conn->call = *call;
428	conn->wdata.active = 1; /* We will activate it asap */	436	conn->wdata.active = 1; /* We will activate it asap */
429	conn->cthread = cthread;	437	conn->cfibril = cfibril;
430		438
431	conn->wdata.ptid = psthread_create(connection_thread, conn);	439	conn->wdata.fid = fibril_create(connection_fibril, conn);
432	if (!conn->wdata.ptid) {	440	if (!conn->wdata.fid) {
433	free(conn);	441	free(conn);
434	ipc_answer_fast(callid, ENOMEM, 0, 0);	442	ipc_answer_fast(callid, ENOMEM, 0, 0);
435	return NULL;	443	return NULL;
436	}	444	}
437	/* Add connection to hash table */	445	/* Add connection to hash table */
438	key = conn->in_phone_hash;	446	key = conn->in_phone_hash;
439	futex_down(&async_futex);	447	futex_down(&async_futex);
440	hash_table_insert(&conn_hash_table, &key, &conn->link);	448	hash_table_insert(&conn_hash_table, &key, &conn->link);
441	futex_up(&async_futex);	449	futex_up(&async_futex);
442		450
443	psthread_add_ready(conn->wdata.ptid);	451	fibril_add_ready(conn->wdata.fid);
444		452
445	return conn->wdata.ptid;	453	return conn->wdata.fid;
446	}	454	}
447		455
448	/** Handle call that was received */	456	/** Handle call that was received */
449	static void handle_call(ipc_callid_t callid, ipc_call_t *call)	457	static void handle_call(ipc_callid_t callid, ipc_call_t *call)
450	{	458	{
Line 457...	Line 465...
457	}	465	}
458		466
459	switch (IPC_GET_METHOD(*call)) {	467	switch (IPC_GET_METHOD(*call)) {
460	case IPC_M_CONNECT_ME_TO:	468	case IPC_M_CONNECT_ME_TO:
461	/* Open new connection with thread etc. */	469	/* Open new connection with thread etc. */
462	async_new_connection(IPC_GET_ARG3(*call), callid, call, client_connection);	470	async_new_connection(IPC_GET_ARG3(*call), callid, call,
-		471	client_connection);
463	return;	472	return;
464	}	473	}
465		474
466	/* Try to route call through connection tables */	475	/* Try to route call through connection tables */
467	if (route_call(callid, call))	476	if (route_call(callid, call))
Line 483...	Line 492...
483	gettimeofday(&tv,NULL);	492	gettimeofday(&tv,NULL);
484	futex_down(&async_futex);	493	futex_down(&async_futex);
485		494
486	cur = timeout_list.next;	495	cur = timeout_list.next;
487	while (cur != &timeout_list) {	496	while (cur != &timeout_list) {
488	waiter = list_get_instance(cur,awaiter_t,link);	497	waiter = list_get_instance(cur, awaiter_t, link);
489	if (tv_gt(&waiter->expires, &tv))	498	if (tv_gt(&waiter->expires, &tv))
490	break;	499	break;
491	cur = cur->next;	500	cur = cur->next;
492	list_remove(&waiter->link);	501	list_remove(&waiter->link);
493	waiter->inlist = 0;	502	waiter->inlist = 0;
Line 495...	Line 504...
495	/* Redundant condition? The thread should not	504	/* Redundant condition? The thread should not
496	* be active when it gets here.	505	* be active when it gets here.
497	*/	506	*/
498	if (!waiter->active) {	507	if (!waiter->active) {
499	waiter->active = 1;	508	waiter->active = 1;
500	psthread_add_ready(waiter->ptid);	509	fibril_add_ready(waiter->fid);
501	}	510	}
502	}	511	}
503		512
504	futex_up(&async_futex);	513	futex_up(&async_futex);
505	}	514	}
Line 512...	Line 521...
512	int timeout;	521	int timeout;
513	awaiter_t *waiter;	522	awaiter_t *waiter;
514	struct timeval tv;	523	struct timeval tv;
515		524
516	while (1) {	525	while (1) {
517	if (psthread_schedule_next_adv(PS_FROM_MANAGER)) {	526	if (fibril_schedule_next_adv(FIBRIL_FROM_MANAGER)) {
518	futex_up(&async_futex);	527	futex_up(&async_futex);
519	/* async_futex is always held	528	/* async_futex is always held
520	* when entering manager thread	529	* when entering manager thread
521	*/	530	*/
522	continue;	531	continue;
523	}	532	}
524	futex_down(&async_futex);	533	futex_down(&async_futex);
525	if (!list_empty(&timeout_list)) {	534	if (!list_empty(&timeout_list)) {
526	waiter = list_get_instance(timeout_list.next,awaiter_t,link);	535	waiter = list_get_instance(timeout_list.next, awaiter_t,
-		536	link);
527	gettimeofday(&tv,NULL);	537	gettimeofday(&tv, NULL);
528	if (tv_gteq(&tv, &waiter->expires)) {	538	if (tv_gteq(&tv, &waiter->expires)) {
529	futex_up(&async_futex);	539	futex_up(&async_futex);
530	handle_expired_timeouts();	540	handle_expired_timeouts();
531	continue;	541	continue;
532	} else	542	} else
Line 570...	Line 580...
570	}	580	}
571		581
572	/** Add one manager to manager list */	582	/** Add one manager to manager list */
573	void async_create_manager(void)	583	void async_create_manager(void)
574	{	584	{
575	pstid_t ptid;	585	fid_t fid;
576		586
577	ptid = psthread_create(async_manager_thread, NULL);	587	fid = fibril_create(async_manager_thread, NULL);
578	psthread_add_manager(ptid);	588	fibril_add_manager(fid);
579	}	589	}
580		590
581	/** Remove one manager from manager list */	591	/** Remove one manager from manager list */
582	void async_destroy_manager(void)	592	void async_destroy_manager(void)
583	{	593	{
584	psthread_remove_manager();	594	fibril_remove_manager();
585	}	595	}
586		596
587	/** Initialize internal structures needed for async manager */	597	/** Initialize internal structures needed for async manager */
588	int _async_init(void)	598	int _async_init(void)
589	{	599	{
590	if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1, &conn_hash_table_ops)) {	600	if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,
-		601	&conn_hash_table_ops)) {
591	printf("%s: cannot create hash table\n", "async");	602	printf("%s: cannot create hash table\n", "async");
592	return ENOMEM;	603	return ENOMEM;
593	}	604	}
594		605
595	return 0;	606	return 0;
596	}	607	}
597		608
598	/** IPC handler for messages in async framework	609	/** IPC handler for messages in async framework
599	*	610	*
600	* Notify thread that is waiting for this message, that it arrived	611	* Notify the fibril which is waiting for this message, that it arrived
601	*/	612	*/
602	static void reply_received(void *private, int retval,	613	static void reply_received(void *private, int retval,
603	ipc_call_t *data)	614	ipc_call_t *data)
604	{	615	{
605	amsg_t msg = (amsg_t ) private;	616	amsg_t msg = (amsg_t ) private;
Line 618...	Line 629...
618	if (msg->wdata.inlist)	629	if (msg->wdata.inlist)
619	list_remove(&msg->wdata.link);	630	list_remove(&msg->wdata.link);
620	msg->done = 1;	631	msg->done = 1;
621	if (! msg->wdata.active) {	632	if (! msg->wdata.active) {
622	msg->wdata.active = 1;	633	msg->wdata.active = 1;
623	psthread_add_ready(msg->wdata.ptid);	634	fibril_add_ready(msg->wdata.fid);
624	}	635	}
625	futex_up(&async_futex);	636	futex_up(&async_futex);
626	}	637	}
627		638
628	/** Send message and return id of the sent message	639	/** Send message and return id of the sent message
Line 634...	Line 645...
634	ipc_call_t *dataptr)	645	ipc_call_t *dataptr)
635	{	646	{
636	amsg_t *msg;	647	amsg_t *msg;
637		648
638	if (in_interrupt_handler) {	649	if (in_interrupt_handler) {
639	printf("Cannot send asynchronous request in interrupt handler.\n");	650	printf("Cannot send asynchronous request in interrupt "
-		651	"handler.\n");
640	_exit(1);	652	_exit(1);
641	}	653	}
642		654
643	msg = malloc(sizeof(*msg));	655	msg = malloc(sizeof(*msg));
644	msg->done = 0;	656	msg->done = 0;
645	msg->dataptr = dataptr;	657	msg->dataptr = dataptr;
646		658
647	msg->wdata.active = 1; /* We may sleep in next method, but it	659	msg->wdata.active = 1; /* We may sleep in next method, but it
648	* will use it's own mechanism */	660	* will use it's own mechanism */
649	ipc_call_async_2(phoneid,method,arg1,arg2,msg,reply_received,1);	661	ipc_call_async_2(phoneid, method, arg1, arg2, msg, reply_received, 1);
650		662
651	return (aid_t) msg;	663	return (aid_t) msg;
652	}	664	}
653		665
654	/** Send message and return id of the sent message	666	/** Send message and return id of the sent message
Line 670...	Line 682...
670	msg->done = 0;	682	msg->done = 0;
671	msg->dataptr = dataptr;	683	msg->dataptr = dataptr;
672		684
673	msg->wdata.active = 1; /* We may sleep in next method, but it	685	msg->wdata.active = 1; /* We may sleep in next method, but it
674	* will use it's own mechanism */	686	* will use it's own mechanism */
675	ipc_call_async_3(phoneid,method,arg1,arg2,arg3, msg,reply_received,1);	687	ipc_call_async_3(phoneid, method, arg1, arg2, arg3, msg, reply_received,
-		688	1);
676		689
677	return (aid_t) msg;	690	return (aid_t) msg;
678	}	691	}
679		692
680	/** Wait for a message sent by async framework	693	/** Wait for a message sent by async framework
Line 692...	Line 705...
692	if (msg->done) {	705	if (msg->done) {
693	futex_up(&async_futex);	706	futex_up(&async_futex);
694	goto done;	707	goto done;
695	}	708	}
696		709
697	msg->wdata.ptid = psthread_get_id();	710	msg->wdata.fid = fibril_get_id();
698	msg->wdata.active = 0;	711	msg->wdata.active = 0;
699	msg->wdata.inlist = 0;	712	msg->wdata.inlist = 0;
700	/* Leave locked async_futex when entering this function */	713	/* Leave locked async_futex when entering this function */
701	psthread_schedule_next_adv(PS_TO_MANAGER);	714	fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
702	/* futex is up automatically after psthread_schedule_next...*/	715	/* futex is up automatically after fibril_schedule_next...*/
703	done:	716	done:
704	if (retval)	717	if (retval)
705	*retval = msg->retval;	718	*retval = msg->retval;
706	free(msg);	719	free(msg);
707	}	720	}
Line 730...	Line 743...
730	}	743	}
731		744
732	gettimeofday(&msg->wdata.expires, NULL);	745	gettimeofday(&msg->wdata.expires, NULL);
733	tv_add(&msg->wdata.expires, timeout);	746	tv_add(&msg->wdata.expires, timeout);
734		747
735	msg->wdata.ptid = psthread_get_id();	748	msg->wdata.fid = fibril_get_id();
736	msg->wdata.active = 0;	749	msg->wdata.active = 0;
737	insert_timeout(&msg->wdata);	750	insert_timeout(&msg->wdata);
738		751
739	/* Leave locked async_futex when entering this function */	752	/* Leave locked async_futex when entering this function */
740	psthread_schedule_next_adv(PS_TO_MANAGER);	753	fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
741	/* futex is up automatically after psthread_schedule_next...*/	754	/* futex is up automatically after fibril_schedule_next...*/
742		755
743	if (!msg->done)	756	if (!msg->done)
744	return ETIMEOUT;	757	return ETIMEOUT;
745		758
746	done:	759	done:
Line 766...	Line 779...
766		779
767	msg = malloc(sizeof(*msg));	780	msg = malloc(sizeof(*msg));
768	if (!msg)	781	if (!msg)
769	return;	782	return;
770		783
771	msg->wdata.ptid = psthread_get_id();	784	msg->wdata.fid = fibril_get_id();
772	msg->wdata.active = 0;	785	msg->wdata.active = 0;
773		786
774	gettimeofday(&msg->wdata.expires, NULL);	787	gettimeofday(&msg->wdata.expires, NULL);
775	tv_add(&msg->wdata.expires, timeout);	788	tv_add(&msg->wdata.expires, timeout);
776		789
777	futex_down(&async_futex);	790	futex_down(&async_futex);
778	insert_timeout(&msg->wdata);	791	insert_timeout(&msg->wdata);
779	/* Leave locked async_futex when entering this function */	792	/* Leave locked async_futex when entering this function */
780	psthread_schedule_next_adv(PS_TO_MANAGER);	793	fibril_schedule_next_adv(FIBRIL_TO_MANAGER);
781	/* futex is up automatically after psthread_schedule_next...*/	794	/* futex is up automatically after fibril_schedule_next...*/
782	free(msg);	795	free(msg);
783	}	796	}
784		797
785	/** Set function that is called, IPC_M_CONNECT_ME_TO is received	798	/** Set function that is called, IPC_M_CONNECT_ME_TO is received
786	*	799	*
Line 797...	Line 810...
797		810
798	/* Primitive functions for simple communication */	811	/* Primitive functions for simple communication */
799	void async_msg_3(int phoneid, ipcarg_t method, ipcarg_t arg1,	812	void async_msg_3(int phoneid, ipcarg_t method, ipcarg_t arg1,
800	ipcarg_t arg2, ipcarg_t arg3)	813	ipcarg_t arg2, ipcarg_t arg3)
801	{	814	{
802	ipc_call_async_3(phoneid, method, arg1, arg2, arg3, NULL, NULL, !in_interrupt_handler);	815	ipc_call_async_3(phoneid, method, arg1, arg2, arg3, NULL, NULL,
-		816	!in_interrupt_handler);
803	}	817	}
804		818
805	void async_msg_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2)	819	void async_msg_2(int phoneid, ipcarg_t method, ipcarg_t arg1, ipcarg_t arg2)
806	{	820	{
807	ipc_call_async_2(phoneid, method, arg1, arg2, NULL, NULL, !in_interrupt_handler);	821	ipc_call_async_2(phoneid, method, arg1, arg2, NULL, NULL,
-		822	!in_interrupt_handler);
808	}	823	}
809		824
810	/** @}	825	/** @}
811	*/	826	*/

Subversion Repositories HelenOS

(root)/trunk/uspace/lib/libc/generic/async.c – Rev 2479 → 2482