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



 
#include <futex.h>
#include <async.h>
#include <fibril.h>
#include <stdio.h>
#include <adt/hash_table.h>
#include <adt/list.h>
#include <ipc/ipc.h>
#include <assert.h>
#include <errno.h>
#include <sys/time.h>
#include <arch/barrier.h>

    /** Fibril function that will be used to handle the connection. */
    void (*cfibril)(ipc_callid_t, ipc_call_t *);
} connection_t;
 
/** Identifier of the incoming connection handled by the current fibril. */
fibril_local connection_t *FIBRIL_connection;
 
static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);
static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call);
 
/**

 * Pointer to a fibril function that will be used to handle interrupt
 * notifications.
 */
static async_client_conn_t interrupt_received = default_interrupt_received;
 
 
static hash_table_t conn_hash_table;
static LIST_INITIALIZE(timeout_list);
 
 
#define CONN_HASH_TABLE_CHAINS  32
 
/** Compute hash into the connection hash table based on the source phone hash.
 *
 * @param key Pointer to source 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_t *msg;
 
       
        msg = list_get_instance(FIBRIL_connection->msg_queue.next,
            msg_t, link);
        list_remove(&msg->link);
        ipc_answer_0(msg->callid, EHANGUP);
        free(msg);
    }
   

            ipc_answer_0(callid, ENOMEM);
        return NULL;
    }
   
    /* Add connection to the connection hash table */
    unsigned long key = conn->in_phone_hash;
   
    futex_down(&async_futex);
    hash_table_insert(&conn_hash_table, &key, &conn->link);
    futex_up(&async_futex);
   

static void handle_call(ipc_callid_t callid, ipc_call_t *call)
{
    /* Unrouted call - do some default behaviour */
    if ((callid & IPC_CALLID_NOTIFICATION)) {
        process_notification(callid, call);
        goto out;
    }
   
    switch (IPC_GET_METHOD(*call)) {
    case IPC_M_CONNECT_ME:
    case IPC_M_CONNECT_ME_TO:
        /* Open new connection with fibril etc. */
        async_new_connection(IPC_GET_ARG5(*call), callid, call,
            client_connection);
        goto out;
    }
   
    /* Try to route the call through the connection hash table */
    if (route_call(callid, call))
        goto out;
   
    /* Unknown call from unknown phone - hang it up */
    ipc_answer_0(callid, EHANGUP);
    return;
   
out:
    ;
}
 
/** Fire all timeouts that expired. */
static void handle_expired_timeouts(void)
{

       
        futex_down(&async_futex);
       
        suseconds_t timeout;
        if (!list_empty(&timeout_list)) {
            awaiter_t *waiter = list_get_instance(timeout_list.next,
                awaiter_t, link);
           
            struct timeval tv;
            gettimeofday(&tv, NULL);
           
            if (tv_gteq(&tv, &waiter->expires)) {

            timeout = SYNCH_NO_TIMEOUT;
       
        futex_up(&async_futex);
       
        ipc_call_t call;
        ipc_callid_t callid = ipc_wait_cycle(&call, timeout,
            SYNCH_FLAGS_NONE);
       
        if (!callid) {
            handle_expired_timeouts();
            continue;
        }

 
/** Initialize the async framework.
 *
 * @return Zero on success or an error code.
 */
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;

 * @param retval Value returned in the answer.
 * @param data   Call data of the answer.
 */
static void reply_received(void *arg, int retval, ipc_call_t *data)
{
    futex_down(&async_futex);
   
    amsg_t *msg = (amsg_t *) arg;
    msg->retval = retval;
   
 
 
    /* Copy data after futex_down, just in case the call was detached */
    if ((msg->dataptr) && (data))
        *msg->dataptr = *data;
   
    write_barrier();
   
    /* Remove message from timeout list */

        return 0;
   
    msg->done = false;
    msg->dataptr = dataptr;
   
    msg->wdata.inlist = false;
    /* We may sleep in the next method, but it will use its own mechanism */
    msg->wdata.active = true;
   
    ipc_call_async_4(phoneid, method, arg1, arg2, arg3, arg4, msg,
        reply_received, true);

        return 0;
   
    msg->done = false;
    msg->dataptr = dataptr;
   
    msg->wdata.inlist = false;
    /* We may sleep in next method, but it will use its own mechanism */
    msg->wdata.active = true;
   
    ipc_call_async_5(phoneid, method, arg1, arg2, arg3, arg4, arg5, msg,
        reply_received, true);

Rev 4377	Rev 4692
Line 93...	Line 93...
93		93
94	#include <futex.h>	94	#include <futex.h>
95	#include <async.h>	95	#include <async.h>
96	#include <fibril.h>	96	#include <fibril.h>
97	#include <stdio.h>	97	#include <stdio.h>
98	#include <libadt/hash_table.h>	98	#include <adt/hash_table.h>
99	#include <libadt/list.h>	99	#include <adt/list.h>
100	#include <ipc/ipc.h>	100	#include <ipc/ipc.h>
101	#include <assert.h>	101	#include <assert.h>
102	#include <errno.h>	102	#include <errno.h>
103	#include <sys/time.h>	103	#include <sys/time.h>
104	#include <arch/barrier.h>	104	#include <arch/barrier.h>
Line 172...	Line 172...
172	/** Fibril function that will be used to handle the connection. */	172	/** Fibril function that will be used to handle the connection. */
173	void (cfibril)(ipc_callid_t, ipc_call_t );	173	void (cfibril)(ipc_callid_t, ipc_call_t );
174	} connection_t;	174	} connection_t;
175		175
176	/** Identifier of the incoming connection handled by the current fibril. */	176	/** Identifier of the incoming connection handled by the current fibril. */
177	__thread connection_t *FIBRIL_connection;	177	fibril_local connection_t *FIBRIL_connection;
178		178
179	static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);	179	static void default_client_connection(ipc_callid_t callid, ipc_call_t *call);
180	static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call);	180	static void default_interrupt_received(ipc_callid_t callid, ipc_call_t *call);
181		181
182	/**	182	/**
Line 188...	Line 188...
188	* Pointer to a fibril function that will be used to handle interrupt	188	* Pointer to a fibril function that will be used to handle interrupt
189	* notifications.	189	* notifications.
190	*/	190	*/
191	static async_client_conn_t interrupt_received = default_interrupt_received;	191	static async_client_conn_t interrupt_received = default_interrupt_received;
192		192
193		-
194	static hash_table_t conn_hash_table;	193	static hash_table_t conn_hash_table;
195	static LIST_INITIALIZE(timeout_list);	194	static LIST_INITIALIZE(timeout_list);
196		195
197		-
198	#define CONN_HASH_TABLE_CHAINS 32	196	#define CONN_HASH_TABLE_CHAINS 32
199		197
200	/** Compute hash into the connection hash table based on the source phone hash.	198	/** Compute hash into the connection hash table based on the source phone hash.
201	*	199	*
202	* @param key Pointer to source phone hash.	200	* @param key Pointer to source phone hash.
Line 498...	Line 496...
498	hash_table_remove(&conn_hash_table, &key, 1);	496	hash_table_remove(&conn_hash_table, &key, 1);
499	futex_up(&async_futex);	497	futex_up(&async_futex);
500		498
501	/* Answer all remaining messages with EHANGUP */	499	/* Answer all remaining messages with EHANGUP */
502	while (!list_empty(&FIBRIL_connection->msg_queue)) {	500	while (!list_empty(&FIBRIL_connection->msg_queue)) {
503	msg_t *msg	501	msg_t *msg;
504	= list_get_instance(FIBRIL_connection->msg_queue.next, msg_t, link);	-
505		502
-		503	msg = list_get_instance(FIBRIL_connection->msg_queue.next,
-		504	msg_t, link);
506	list_remove(&msg->link);	505	list_remove(&msg->link);
507	ipc_answer_0(msg->callid, EHANGUP);	506	ipc_answer_0(msg->callid, EHANGUP);
508	free(msg);	507	free(msg);
509	}	508	}
510		509
Line 561...	Line 560...
561	ipc_answer_0(callid, ENOMEM);	560	ipc_answer_0(callid, ENOMEM);
562	return NULL;	561	return NULL;
563	}	562	}
564		563
565	/* Add connection to the connection hash table */	564	/* Add connection to the connection hash table */
566	ipcarg_t key = conn->in_phone_hash;	565	unsigned long key = conn->in_phone_hash;
567		566
568	futex_down(&async_futex);	567	futex_down(&async_futex);
569	hash_table_insert(&conn_hash_table, &key, &conn->link);	568	hash_table_insert(&conn_hash_table, &key, &conn->link);
570	futex_up(&async_futex);	569	futex_up(&async_futex);
571		570
Line 586...	Line 585...
586	static void handle_call(ipc_callid_t callid, ipc_call_t *call)	585	static void handle_call(ipc_callid_t callid, ipc_call_t *call)
587	{	586	{
588	/* Unrouted call - do some default behaviour */	587	/* Unrouted call - do some default behaviour */
589	if ((callid & IPC_CALLID_NOTIFICATION)) {	588	if ((callid & IPC_CALLID_NOTIFICATION)) {
590	process_notification(callid, call);	589	process_notification(callid, call);
591	return;	590	goto out;
592	}	591	}
593		592
594	switch (IPC_GET_METHOD(*call)) {	593	switch (IPC_GET_METHOD(*call)) {
-		594	case IPC_M_CONNECT_ME:
595	case IPC_M_CONNECT_ME_TO:	595	case IPC_M_CONNECT_ME_TO:
596	/* Open new connection with fibril etc. */	596	/* Open new connection with fibril etc. */
597	async_new_connection(IPC_GET_ARG5(*call), callid, call,	597	async_new_connection(IPC_GET_ARG5(*call), callid, call,
598	client_connection);	598	client_connection);
599	return;	599	goto out;
600	}	600	}
601		601
602	/* Try to route the call through the connection hash table */	602	/* Try to route the call through the connection hash table */
603	if (route_call(callid, call))	603	if (route_call(callid, call))
604	return;	604	goto out;
605		605
606	/* Unknown call from unknown phone - hang it up */	606	/* Unknown call from unknown phone - hang it up */
607	ipc_answer_0(callid, EHANGUP);	607	ipc_answer_0(callid, EHANGUP);
-		608	return;
-		609
-		610	out:
-		611	;
608	}	612	}
609		613
610	/** Fire all timeouts that expired. */	614	/** Fire all timeouts that expired. */
611	static void handle_expired_timeouts(void)	615	static void handle_expired_timeouts(void)
612	{	616	{
Line 660...	Line 664...
660		664
661	futex_down(&async_futex);	665	futex_down(&async_futex);
662		666
663	suseconds_t timeout;	667	suseconds_t timeout;
664	if (!list_empty(&timeout_list)) {	668	if (!list_empty(&timeout_list)) {
665	awaiter_t *waiter	669	awaiter_t *waiter = list_get_instance(timeout_list.next,
666	= list_get_instance(timeout_list.next, awaiter_t, link);	670	awaiter_t, link);
667		671
668	struct timeval tv;	672	struct timeval tv;
669	gettimeofday(&tv, NULL);	673	gettimeofday(&tv, NULL);
670		674
671	if (tv_gteq(&tv, &waiter->expires)) {	675	if (tv_gteq(&tv, &waiter->expires)) {
Line 678...	Line 682...
678	timeout = SYNCH_NO_TIMEOUT;	682	timeout = SYNCH_NO_TIMEOUT;
679		683
680	futex_up(&async_futex);	684	futex_up(&async_futex);
681		685
682	ipc_call_t call;	686	ipc_call_t call;
683	ipc_callid_t callid	687	ipc_callid_t callid = ipc_wait_cycle(&call, timeout,
684	= ipc_wait_cycle(&call, timeout, SYNCH_FLAGS_NONE);	688	SYNCH_FLAGS_NONE);
685		689
686	if (!callid) {	690	if (!callid) {
687	handle_expired_timeouts();	691	handle_expired_timeouts();
688	continue;	692	continue;
689	}	693	}
Line 732...	Line 736...
732		736
733	/** Initialize the async framework.	737	/** Initialize the async framework.
734	*	738	*
735	* @return Zero on success or an error code.	739	* @return Zero on success or an error code.
736	*/	740	*/
737	int _async_init(void)	741	int __async_init(void)
738	{	742	{
739	if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,	743	if (!hash_table_create(&conn_hash_table, CONN_HASH_TABLE_CHAINS, 1,
740	&conn_hash_table_ops)) {	744	&conn_hash_table_ops)) {
741	printf("%s: cannot create hash table\n", "async");	745	printf("%s: cannot create hash table\n", "async");
742	return ENOMEM;	746	return ENOMEM;
Line 756...	Line 760...
756	* @param retval Value returned in the answer.	760	* @param retval Value returned in the answer.
757	* @param data Call data of the answer.	761	* @param data Call data of the answer.
758	*/	762	*/
759	static void reply_received(void arg, int retval, ipc_call_t data)	763	static void reply_received(void arg, int retval, ipc_call_t data)
760	{	764	{
-		765	futex_down(&async_futex);
-		766
761	amsg_t msg = (amsg_t ) arg;	767	amsg_t msg = (amsg_t ) arg;
762	msg->retval = retval;	768	msg->retval = retval;
763		769
764	futex_down(&async_futex);	-
765		-
766	/* Copy data after futex_down, just in case the call was detached */	770	/* Copy data after futex_down, just in case the call was detached */
767	if (msg->dataptr)	771	if ((msg->dataptr) && (data))
768	msg->dataptr = data;	772	msg->dataptr = data;
769		773
770	write_barrier();	774	write_barrier();
771		775
772	/* Remove message from timeout list */	776	/* Remove message from timeout list */
Line 808...	Line 812...
808	return 0;	812	return 0;
809		813
810	msg->done = false;	814	msg->done = false;
811	msg->dataptr = dataptr;	815	msg->dataptr = dataptr;
812		816
-		817	msg->wdata.inlist = false;
813	/* We may sleep in the next method, but it will use its own mechanism */	818	/* We may sleep in the next method, but it will use its own mechanism */
814	msg->wdata.active = true;	819	msg->wdata.active = true;
815		820
816	ipc_call_async_4(phoneid, method, arg1, arg2, arg3, arg4, msg,	821	ipc_call_async_4(phoneid, method, arg1, arg2, arg3, arg4, msg,
817	reply_received, true);	822	reply_received, true);
Line 847...	Line 852...
847	return 0;	852	return 0;
848		853
849	msg->done = false;	854	msg->done = false;
850	msg->dataptr = dataptr;	855	msg->dataptr = dataptr;
851		856
-		857	msg->wdata.inlist = false;
852	/* We may sleep in next method, but it will use its own mechanism */	858	/* We may sleep in next method, but it will use its own mechanism */
853	msg->wdata.active = true;	859	msg->wdata.active = true;
854		860
855	ipc_call_async_5(phoneid, method, arg1, arg2, arg3, arg4, arg5, msg,	861	ipc_call_async_5(phoneid, method, arg1, arg2, arg3, arg4, arg5, msg,
856	reply_received, true);	862	reply_received, true);

Subversion Repositories HelenOS

(root)//branches/tracing/uspace/lib/libc/generic/async.c – Rev 4377 → 4692