# HG changeset patch # User kent # Date 1256119390 -32400 # Node ID 0c614303d3059b7c94f31fe48f8f79b14b78ee60 # Parent f8d56635efb80f4b3faeacbbc02bc860f06cf143 add new lindaapi into test_render/lindaapi diff -r f8d56635efb8 -r 0c614303d305 TaskManager/Test/test_render/lindaapi/Makefile --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/TaskManager/Test/test_render/lindaapi/Makefile Wed Oct 21 19:03:10 2009 +0900 @@ -0,0 +1,23 @@ +CC=gcc +AR=ar +CFLAGS=-g -O2 -Wall +#FLAGS=-Wall -DDEBUG +FLAGS=-Wall +TARGET=liblindaapi.a +LINDADIR=. +LIBDIR=$(LINDADIR) + +all : $(TARGET) + +ldserv : $(LINDADIR)/ldserv.c + $(CC) $(FLAGS) -g -o ldserv $(LINDADIR)/ldserv.c + +.c.o: + $(CC) $(CFLAGS) -c $^ -o $@ + +liblindaapi.a : lindaapi.o + $(AR) crus $@ $^ + +clean : + rm -f $(TARGET) *.o + diff -r f8d56635efb8 -r 0c614303d305 TaskManager/Test/test_render/lindaapi/lindaapi.c --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/TaskManager/Test/test_render/lindaapi/lindaapi.c Wed Oct 21 19:03:10 2009 +0900 @@ -0,0 +1,741 @@ +// $Id$ +// + +/*---------------------------------------------------------------------- + インクルードファイル読み込み +----------------------------------------------------------------------*/ +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "lindaapi.h" + + +#if 0 +#define PSX_Debug(deb) (putchar(PS_DEB)),\ + (printf deb ),\ + (putchar(PS_DEB)) +#define DEB(a) +#else +#define PSX_Debug(deb) +#define DEB(a) /* a */ +#endif + +/* Global Variables */ +static COMMAND *q_top, *q_end; /* コマンドキュー */ +static REPLY *reply, *r_end; /* 受け取り用キュー */ +static int qsize; /* コマンドキューのサイズ */ +static fd_set g_fds; /* 接続しているタプルスペース群のFD(FileDiscripter)を保持 */ +static unsigned int g_max_fds; /* 監視するFDの最大値 */ + +/* Static Functions */ +static void unix_chkserv(int ps); +void psx_free(void *); +static int psx_queue(unsigned int tspace_id, unsigned int id, + unsigned int size, unsigned char *data, char mode, + void(*callback)(unsigned char *,void *),void * obj); + +#ifdef COUNT_PACKET +// print packet count message per PRINT_INTERVAL sec +#define PRINT_INTERVAL 4 +static void count_packet(char type); + +/*-------------------------------------------------------------------/ + static void + count_packet (char type): + パケットの送受信カウントする + + 引き数: + type - 送信、受信 (char型: s,r) +/-------------------------------------------------------------------*/ +static void +count_packet(char type) +{ + static int send_packet=-1,receive_packet=0; + static struct timeval start,now,previous; + + if (out_packet == -1) { + gettimeofday(&start,NULL); + gettimeofday(&previous,NULL); + send_packet = 0; + printf("packet\tout\tread\t\ttime\n"); + } + + if (type == 's') { + send_packet++; + } else if (type == 'r') { + receive_packet++; + } else { + fprintf(stderr,"No type in count_packet function\n"); + return; + } + + gettimeofday(&now,NULL); + if ((now.tv_sec-previous.tv_sec) > PRINT_INTERVAL) { + printf("log\t%d\t%d\t%ld\n", + send_packet,receive_packet,now.tv_sec-start.tv_sec); + fflush(stdout); + + previous.tv_sec = now.tv_sec; + send_packet = receive_packet = 0; + } +} +#endif + + +#define unix_read_w read + +/*-------------------------------------------------------------------/ + static int + unix_write (int fd, unsigned char *buf, unsigned int size): + サーバへTUPLEを送る。 + + 引き数: + fd - サーバのファイルディスクリプタ + buf - サーバへ送るデータ(TUPLEヘッダ含む) + size - bufのbyte数 + 返り値: + 送った(書きこんだ)データのbyte数 +/-------------------------------------------------------------------*/ +static int +unix_write_bak(int fd,unsigned char *buf,unsigned int size) { + int i,nsize; + nsize = htonl(size); + i = write(fd,&nsize,INT_SIZE); + i += write(fd,buf,size); // size == datasize + LINDA_HEADER_SIZE +#ifdef COUNT_PACKET + count_packet('s'); +#endif + return(i); +} +static int +unix_write(int fd,unsigned char *buf,unsigned int size) { + int count=0; + uint32_t nsize; + + /* これから送信するデータのサイズをまず送信 */ + nsize = htonl(size); + write(fd, &nsize, INT_SIZE); + + /* 目的のデータを送信 */ + while (count < size) { + count += write(fd, buf+count, size-count); + } +#ifdef COUNT_PACKET + count_packet('s'); +#endif + return count+INT_SIZE; +} + +#define unix_write_w unix_write + +#define SERV_NAME unix_port +#define PROTO_NAME "tcp" +#define SERVER_NAME hostname +#define MAX_REQ 16 + + + +/*-------------------------------------------------------------------/ + void + init_linda(): + 大域変数の初期化等を行なう +/-------------------------------------------------------------------*/ +void +init_linda() { + FD_ZERO(&g_fds); + /* 大域変数はゼロクリアされる + g_max_fds = 0; + q_end = q_top = NULL; + r_end = reply = NULL; + qsize = 0; + */ +} + + +/*-------------------------------------------------------------------/ + int + open_linda (char * hostname, int port): + Lindaサーバとのコネクションを確立し、タプルスペースのIDを返す。 + 現在はファイルディスクリプタを返している。 + + 引き数: + hostname - サーバのホスト名 + port - サーバのポート番号 + 返り値: + コネクション確立が成功するとそのファイルディスクリプタを返す。 + 失敗すると -1 を返す。 +/-------------------------------------------------------------------*/ +int +open_linda(char * hostname, int port){ + int fd; + struct hostent *hoste; + struct sockaddr_in serv_addr; + struct sockaddr_un serv_addr_un; + + if (hostname[0]=='/') { + /* Unix domain */ + if ((fd = socket(AF_UNIX, SOCK_STREAM, 0)) == FAIL){ + perror("socket"); + return(-1); + } + serv_addr_un.sun_family = AF_UNIX; + strcpy(serv_addr_un.sun_path, hostname); + fprintf(stdout,"connecting ... %d\n", serv_addr.sin_port); + if (connect(fd, (struct sockaddr *)&serv_addr_un,sizeof(serv_addr_un)) == FAIL){ + perror("connect"); + close(fd); + return(-1); + } + + } else { + /* INET domain */ + if ((fd = socket(AF_INET, SOCK_STREAM, 0)) == FAIL){ + perror("socket"); + return(-1); + } + if ((hoste = gethostbyname(SERVER_NAME)) == NULL){ + fprintf(stderr,"hostname error\n"); + close(fd); + return(-1); + } + serv_addr.sin_family = AF_INET; + serv_addr.sin_port = port; + serv_addr.sin_addr.s_addr = ((struct in_addr *)(hoste->h_addr))->s_addr; + if (serv_addr.sin_family == AF_INET) { + int tmp = 1; + setsockopt (fd, IPPROTO_TCP, TCP_NODELAY, + (char *) &tmp, sizeof (int)); + } + fprintf(stdout,"connecting ... %d\n", serv_addr.sin_port); + if (connect(fd, (struct sockaddr *)&serv_addr,sizeof(serv_addr)) == FAIL){ + fprintf(stderr,"connection error! errno :%d %s\n", errno, + strerror(errno)); + close(fd); + return(-1); + } + } + + FD_SET(fd, &g_fds); + if (g_max_fds < fd) g_max_fds = fd; + + fprintf(stdout," connect middle server %d\n", fd); + return fd; +} + +int +open_linda_java(char * hostname, int port){ + int fd; + struct hostent *hoste; + struct sockaddr_in serv_addr; + struct sockaddr_un serv_addr_un; + + if (hostname[0]=='/') { + /* Unix domain */ + if ((fd = socket(AF_UNIX, SOCK_STREAM, 0)) == FAIL){ + perror("socket"); + return(-1); + } + serv_addr_un.sun_family = AF_UNIX; + strcpy(serv_addr_un.sun_path, hostname); + DEB(fprintf(stdout,"connecting ... %d\n", serv_addr.sin_port)); + if (connect(fd, (struct sockaddr *)&serv_addr_un,sizeof(serv_addr_un)) == FAIL){ + perror("connect"); + close(fd); + return(-1); + } + + } else { + /* INET domain */ + if ((fd = socket(AF_INET, SOCK_STREAM, 0)) == FAIL){ + perror("socket"); + return(-2); + } + serv_addr.sin_family = AF_INET; + serv_addr.sin_port = htons(port); + + serv_addr.sin_addr.s_addr = inet_addr(hostname); + if (serv_addr.sin_addr.s_addr == 0xffffffff) { + if ((hoste = gethostbyname(hostname)) == NULL){ + fprintf(stdout, "hostname error\n"); + close(fd); + return(-1); + } + serv_addr.sin_addr.s_addr = *(unsigned int *)hoste->h_addr_list[0]; + } + + if (serv_addr.sin_family == AF_INET) { + int tmp = 1; + setsockopt (fd, IPPROTO_TCP, TCP_NODELAY, + (char *) &tmp, sizeof (int)); + } + DEB(fprintf(stdout,"connecting ... %d \n", ntohs(serv_addr.sin_port))); + DEB(fprintf(stdout," serv_addr.sin_port ... %d \n", ntohs(serv_addr.sin_port))); + //fprintf(stdout," serv_addr.sin_addr.s_addr... %s\n", serv_addr.sin_addr.s_addr); + if (connect(fd, (struct sockaddr *)&serv_addr,sizeof(serv_addr)) == FAIL){ + perror("connect"); + close(fd); + return(-4); + } + } + + FD_SET(fd, &g_fds); + if (g_max_fds < fd) g_max_fds = fd; + + DEB(fprintf(stdout," connect middle server %d\n", fd)); + return fd; +} + + +/*-------------------------------------------------------------------/ + int + close_linda(int tspace_id): + 接続しているタプルスペースへの接続を切る。 + ソケットを閉じ、g_fds から外す。 + 引数: + tspace_id - 閉じるタプルスペースのID + 返り値: + close の値 +/-------------------------------------------------------------------*/ +int +close_linda(int tspace_id){ + int retval; + int i; + if ((retval = close(tspace_id)) == 0) { + FD_CLR(tspace_id, &g_fds); + if (g_max_fds == tspace_id) { + for (i = g_max_fds-1; FD_ISSET(i, &g_fds) && i; i--); + g_max_fds = i; + } + } + return retval; +} + +/*-------------------------------------------------------------------/ + int + psx_out (unsigned int tspace_id, unsigned int id, + unsigned char *data, unsigned int size): + outコマンドをCOMMANDキューへ溜める。 + + 引き数: + tspace_id - タプルスペースのID + id - タプルのID + data - 送信するデータ + size - dataのサイズ + 返り値: + シーケンス番号 +/-------------------------------------------------------------------*/ +int +psx_out(unsigned int tspace_id, unsigned int id, + unsigned char *data, unsigned int size){ + unsigned int r; + if ((r = psx_queue(tspace_id, id, size, data, 'o', NULL, NULL)) == FAIL) { + return(FAIL); + } + DEB( fprintf(stdout, "psx_out: size = %d, command = %s\n", + q_end->size, q_end->command+LINDA_HEADER_SIZE)); + return(r); +} + +/*-------------------------------------------------------------------/ + int + psx_ld (unsigned tspace_id, unsigned int id, + char mode, void(*callback)(char*,void*), void * obj): + in,read,waitなどの受信コマンドをCOMMANDキューへ溜める。 + psx_in,psx_rd,psx_wait_rdなどに置き換えられている。 + + 引き数: + tspace_id- タプルスペースのID + id - タプルのID + mode - i,r,w の文字を取り、各々in,read,waitを表している。 + callback - コールバックを使用する場合の関数へのポインタ。 + 使用しない場合はNULLをいれる。 + obj - コールバックで用いる関数の引き数。 + 返り値: + psx_queue内でmallocされたREPLY構造体へのポインタ +/-------------------------------------------------------------------*/ +int +psx_ld(unsigned int tspace_id, unsigned int id, + char mode, void(*callback)(unsigned char *,void *), void * obj){ + int r; + if ((r = psx_queue(tspace_id, id, 0, NULL, mode, callback, obj)) == FAIL) { + return(FAIL); + } + return(r); +} + +/*-------------------------------------------------------------------/ + unsigned char * + psx_reply (int seq): + サーバから答えが来たデータを返す。 + + 引き数: + seq - psx_ld()が返した値。 + 返り値: + seqに対応したデータを返す。データをまだ受信していない場合は + NULLを返す。 +/-------------------------------------------------------------------*/ +unsigned char * +psx_reply(int seq){ + REPLY *p, *q; + unsigned char *ans; + + DEB(fprintf(stdout, "psx_reply: search of seq = %d\n", seq)); + PSX_Debug(("psx_reply: seq %d", seq)); + for(q = NULL,p = reply; p; q = p,p = p->next){ + if (p->seq == seq){ + DEB(fprintf(stdout, "psx_reply: match of seq = %d\n", seq)); + if (p->mode == '!'){ + ans = p->answer; + if (q == NULL){ + reply = p->next; + if(p==r_end) { + r_end = p->next; + } + } else { + q->next = p->next; + if(p==r_end) { + r_end = q; + } + } + PSX_Debug(("psx_reply: reply %x r_end %x p %x q %x",reply,r_end,p,q)); + psx_free(p); + DEB( for(p=reply;p;p=p->next) { PSX_Debug(("psx_queue dump: seq %d mode %c %x %x",p->seq,p->mode,p,p->next))}); + DEB( fprintf(stdout, "psx_reply: returned answer = %s\n", ans)); + PSX_Debug(("psx_reply: answer %s",ans)); + return(ans); + } else { + if (p->mode == '?'){ + DEB(fprintf(stdout, "psx_reply: don't accept anser\n")); + return(NULL); + } + } + } + + } + PSX_Debug(("psx_reply: no match seq %d",seq)); + DEB(fprintf(stdout, "psx_reply: no match of seq\n")); + return(NULL); +} + +/*-------------------------------------------------------------------/ + void + psx_sync_n (): + サーバとデータの送受信をする。COMMANDキューに溜まったデータを + 送信し、サーバから送られて来たデータを対応するREPLYへいれる。 +/-------------------------------------------------------------------*/ +#define TIMEDELTA 10 +void +psx_sync_n(){ + int acount; + int i; + COMMAND *c, *t; + + fd_set tmp; + struct timeval timeout; + timeout.tv_sec=0; + timeout.tv_usec=TIMEDELTA * 1000; + + acount = 0; + while (q_top != NULL){ + c = q_top; + unix_write_w(c->tspace_id, c->command, c->size); + psx_free(c->command); + t = c->next; + psx_free(c); + q_top = c = t; + qsize--; + } + + tmp = g_fds; + while(select(g_max_fds+1, &tmp, NULL, NULL, &timeout) > 0) { + for (i = 0; i < g_max_fds+1; i++) { + if (FD_ISSET(i, &tmp)) { + unix_chkserv(i); + } + } + } +} + +/*-------------------------------------------------------------------/ + static int + psx_queue (unsigned int tspace_id, unsigned int id, + unsigned int size, unsigned char *data, char mode, + void(*callback)(char*,void*), void * obj): + out,in,read,waitなどのコマンドをCOMMANDキューに溜める。データを + 受信するコマンド(in,read,wait)のときは受け取ったときにデータを + 格納するREPLY構造体を作る。 + + 引き数: + tspace_id- 送信先タプルスペースのID + id - アクセスするTUPLE SpaceのID + size - dataのサイズ + data - 送信するデータ。受信時はNULL。 + mode - コマンドのモード(out,in,read,wait は各々char型: o,i,r,w) + callback - コールバックを使用する場合の関数へのポインタ。 + 使用しない場合はNULL。 + obj - コールバックで用いる関数に引き渡すデータ。 + 返り値: + 成功した場合 - mallocしたREPLY構造体へのポインタ。outの場合は + 0が返る。 + 失敗した場合 - FAIL(-1)が返る。 +/-------------------------------------------------------------------*/ +static int +psx_queue(unsigned int tspace_id, unsigned int id, + unsigned int size, unsigned char *data, char mode, + void(*callback)(unsigned char *,void *), void * obj){ + REPLY *p; + COMMAND *c; + + if (qsize >= MAX_QUEUE) { +// PSX_Debug(("max queue: qsize=%d",qsize)); + psx_sync_n(); + } + + if (q_top == NULL) { + if ((q_top = (COMMAND *) malloc (sizeof(COMMAND))) == NULL){ + return(FAIL); + } + c = q_end = q_top; + } else { + if ((q_end->next = (COMMAND *) malloc (sizeof(COMMAND))) == NULL){ + return(FAIL); + } + c = q_end; + q_end = q_end->next; + } + + /* size は DATASIZE */ + if ((q_end->command = (unsigned char *) malloc(size+LINDA_HEADER_SIZE)) == NULL) { + psx_free(q_end); + c->next = NULL; + return(FAIL); + } + + /* データ受け取り要求(in,rd,wait)なら受け取り用の箱を用意 */ + if (mode != 'o') { + if (reply == NULL){ + if ((reply = (REPLY *) malloc (sizeof(REPLY))) == NULL){ + return(FAIL); + } + p = r_end = reply; p->next = NULL; + } else { + if ((r_end->next = (REPLY *) malloc (sizeof(REPLY))) == NULL){ + return(FAIL); + } + p = r_end->next; r_end = p; p->next = NULL; + } + p->mode = '?'; + p->seq = (int)p; // 構造体のアドレスで識別 + p->callback = callback; + p->obj = obj; + PSX_Debug(("psx_queue: seq %d reply %x p %x r_end %x",seq,reply,p,r_end)); + } else { + p = 0; + } + q_end->command[LINDA_MODE_OFFSET] = mode; + + q_end->command[LINDA_ID_OFFSET] = id >> 8; + q_end->command[LINDA_ID_OFFSET+1] = id & 0xff; + + q_end->command[LINDA_SEQ_OFFSET] = ((int)p>>24) & 0xff; + q_end->command[LINDA_SEQ_OFFSET+1] = ((int)p>>16) & 0xff; + q_end->command[LINDA_SEQ_OFFSET+2] = ((int)p>>8) & 0xff; + q_end->command[LINDA_SEQ_OFFSET+3] = ((int)p) & 0xff; + + q_end->command[LINDA_DATA_LENGTH_OFFSET] = (size>>24) & 0xff; + q_end->command[LINDA_DATA_LENGTH_OFFSET+1] = (size>>16) & 0xff; + q_end->command[LINDA_DATA_LENGTH_OFFSET+2] = (size>>8) & 0xff; + q_end->command[LINDA_DATA_LENGTH_OFFSET+3] = (size) & 0xff; + + q_end->size = size+LINDA_HEADER_SIZE; /* command size */ + q_end->tspace_id = tspace_id; /* destination id */ + q_end->next = NULL; + qsize++; + if (data && size>0) + memcpy(q_end->command+LINDA_HEADER_SIZE, data, size); + return((int)p); +} + +/*-------------------------------------------------------------------/ + static void + unix_chkserv (int ps): + サーバからデータ(TUPLE)を受け取る。REPLY構造体にコールバック関数 + が指定されていればその関数を実行し、REPLY構造体をキューから取り + 除く。コールバック関数が指定されていなければREPLY構造体にデータ + を引き渡す。 + 引数: + ps - 接続しているタプルスペースのソケット +/-------------------------------------------------------------------*/ +static void +unix_chkserv(int ps){ + int i,k,pkt,npkt,mode; + REPLY *r,*prev; + int a; + unsigned char * tuple = 0; + + if((i=read(ps,&npkt,INT_SIZE))<0) { + perror("read"); + exit(1); + } + pkt = ntohl(npkt); + DEB(printf("pkt: %d\n",pkt)); + DEB(fprintf(stdout, "psx_chkserv: queue number: %d , size = %d\n", i, pkt)); + if((tuple = (unsigned char *)malloc(pkt))==NULL){ + perror("malloc"); + exit(1); + } + for(a=0;anext) { + PSX_Debug(printf("psx_queue dump: seq %d mode %c %x %x",p->seq,p->mode,p,p->next)); + }) + + for(prev = NULL,r = reply; r; prev = r,r = r->next){ + DEB(fprintf(stdout,"seq: %d\n",r->seq);) + if (r->seq == k){ + if(r->callback){ // call callback function + (*r->callback)(tuple,r->obj); + if (prev == NULL){ + reply = r->next; + if(r == r_end) { + r_end = r->next; + } + } else { + prev->next = r->next; + if(r == r_end) { + r_end = prev; + } + } + psx_free(r); + }else{ // normal reply + PSX_Debug(("psx_chkserv: copy answer r %x seq %d",r,k)); + if(mode == 'a'){ + r->answer = tuple; + }else{ + r->answer = NULL; + } + r->mode = '!'; + } + break; + } + } + tuple = 0; + if (!r){ + DEB(fprintf(stdout, "unix_chkserv: accepted seq %d does not match. \n",k)); + } +} + +void psx_free(void *tuple) +{ + free(tuple); +} + +/*-------------------------------------------------------------------/ + static unsigned int + get_int(unsigned char * tuple, int offset): + TUPLEのヘッダに格納された int型 のデータを得るための関数 + psx_get_datalength() と psx_get_seq() から呼ばれる。 + + 引き数: + tuple - ヘッダ情報も含んだTUPLE。psx_reply()で得たものでもいい。 + offset - 取りだすデータのオフセット。LINDA_DATA_LENGTH_OFFSET + か LINDA_SEQ_OFFSET。 + + 返り値: + 指定したオフセットに格納されていた数値(int型) +/-------------------------------------------------------------------*/ +static unsigned int +get_int(unsigned char * tuple, int offset){ + unsigned int i; + i = (tuple[offset] <<24) + + (tuple[offset+1]<<16) + + (tuple[offset+2]<<8) + + (tuple[offset+3]); + return i; +} + +unsigned int +psx_get_datalength(unsigned char * tuple){ + return get_int(tuple,LINDA_DATA_LENGTH_OFFSET); +} + +unsigned char * +psx_get_data(unsigned char * tuple) { + return tuple + LINDA_HEADER_SIZE; +} + +unsigned int +psx_get_seq(unsigned char * tuple){ + return get_int(tuple,LINDA_SEQ_OFFSET); +} + +unsigned short +psx_get_id(unsigned char * tuple){ + return (tuple[LINDA_ID_OFFSET] * 256 + tuple[LINDA_ID_OFFSET+1]); +} + +unsigned char +psx_get_mode(unsigned char * tuple){ + return tuple[LINDA_MODE_OFFSET]; +} + +static +void +set_int_to_char(unsigned char * tuple, int i, int offset){ + tuple[offset] = (i>>24) & 0xff; + tuple[offset+1] = (i>>16) & 0xff; + tuple[offset+2] = (i>>8) & 0xff; + tuple[offset+3] = (i) & 0xff; +} + +void +psx_set_datalength(unsigned char * tuple, int length){ + set_int_to_char(tuple,length,LINDA_DATA_LENGTH_OFFSET); +} + + +void +psx_set_seq(unsigned char * tuple, int seq){ + set_int_to_char(tuple,seq,LINDA_SEQ_OFFSET); +} + +void +psx_set_id(unsigned char * tuple, short id){ + tuple[LINDA_ID_OFFSET] = id >> 8; + tuple[LINDA_ID_OFFSET+1] = id & 0xff; +} + +void +psx_set_mode(unsigned char * tuple, char mode){ + tuple[LINDA_MODE_OFFSET] = mode; +} + + + +/* end */ diff -r f8d56635efb8 -r 0c614303d305 TaskManager/Test/test_render/lindaapi/lindaapi.h --- /dev/null Thu Jan 01 00:00:00 1970 +0000 +++ b/TaskManager/Test/test_render/lindaapi/lindaapi.h Wed Oct 21 19:03:10 2009 +0900 @@ -0,0 +1,82 @@ +/* $Id$ */ + +/*---------------------------------------------------------------------- + マクロ定義 +----------------------------------------------------------------------*/ + +#define FAIL (-1) + +#define PS_DEB 0xf8 + +#define MAX_QUEUE 1024 + +#define INT_SIZE 4 // Byte = sizeof(int) + +#define LDSERV_PORT 11511 + +/*---------------------------------------------------------------------- + パケットフォーマット + char short int int + Mode + ID + Seq + Data_len + Padding + Data + 0 1 3 7 11 12 +----------------------------------------------------------------------*/ + +#define LINDA_MODE_OFFSET 0 +#define LINDA_ID_OFFSET 1 +#define LINDA_SEQ_OFFSET 3 +#define LINDA_DATA_LENGTH_OFFSET 7 +#define LINDA_HEADER_SIZE 12 + +/*---------------------------------------------------------------------- + 構造体定義 +----------------------------------------------------------------------*/ +typedef struct psx_reply{ + unsigned char *answer; + unsigned int seq; + struct psx_reply *next; + char mode; + void(*callback)(unsigned char * tuple, void * obj); + void * obj; +} REPLY; + +typedef struct command_queue{ + struct command_queue *next; + unsigned int tspace_id; /* ID of destination Tuple Space */ + unsigned int size; + unsigned char *command; +} COMMAND; + + +/*---------------------------------------------------------------------- + Linda API +----------------------------------------------------------------------*/ + +extern void init_linda(void); +extern int open_linda(char *hostname, int port); +extern int close_linda(int tspace_id); + +extern void psx_free(void*); + +extern unsigned char *psx_reply(int seq); +extern void psx_sync_n(void); +extern unsigned int psx_get_datalength(unsigned char *); +extern unsigned int psx_get_seq(unsigned char *); +extern unsigned short psx_get_id(unsigned char *); +extern unsigned char psx_get_mode(unsigned char *); +extern unsigned char * psx_get_data(unsigned char *); + +extern int psx_out(unsigned int tspace_id, unsigned int id, + unsigned char *data, unsigned int size); +extern int psx_ld(unsigned int tspace_id, unsigned int id, + char mode, void(*callback)(unsigned char *,void *),void * obj); + +#define psx_in(tid, id) psx_ld(tid, id, 'i', NULL, NULL) +#define psx_rd(tid, id) psx_ld(tid, id, 'r', NULL, NULL) +#define psx_ck(tid, id) psx_ld(tid, id, 'c', NULL, NULL) +#define psx_wait_rd(tid, id) psx_ld(tid, id, 'w', NULL, NULL) + +#define psx_callback_in(tid, id,callback,obj) psx_ld(tid, id, 'i', callback, obj) +#define psx_callback_rd(tid, id,callback,obj) psx_ld(tid, id, 'r', callback, obj) +#define psx_callback_ck(tid, id,callback,obj) psx_ld(tid, id, 'c', callback, obj) +#define psx_callback_wait_rd(tid, id,callback,obj) psx_ld(tid, id, 'w', callback, obj) +