code.grnet.gr Git - archipelago/blob - xseg/peers/user/monitor.c

   1 #include <stdio.h>
   2 #include <unistd.h>
   3 #include <sys/types.h>
   4 #include <pthread.h>
   5 #include <xseg/xseg.h>
   6 #include <mpeer.h>
   7 #include <sys/time.h>
   8
   9 #define INPUT_BUF_SIZE 256
  10 #define MAX_NR_ARGS 100
  11
  12 struct monitord {
  13         uint32_t mon_portno;
  14 };
  15
  16 struct monitor_io {
  17         uint32_t src_portno;
  18         void *src_priv;
  19 };
  20
  21
  22
  23 static int forward(struct peerd *peer, struct peer_req *pr)
  24 {
  25         int r;
  26         r = submit_peer_req(peer, pr);
  27         if (r < 0) {
  28                 printf("couldn't forward request");
  29                 return -1;
  30         }
  31         return 0;
  32 }
  33
  34 static int complete_forwarded(struct peerd *peer, struct peer_req *pr)
  35 {
  36         struct xseg_request *req = pr->req;
  37
  38         // assert mio->src_portno != NoPort
  39         if (req->state & XS_SERVED)
  40                 complete(peer, pr);
  41         else if (req->state & XS_FAILED)
  42                 fail (peer, pr);
  43         else {
  44                 printf("invalid state\n");
  45                 return -1;
  46         }
  47         return 0;
  48 }
  49
  50 int dispatch(struct peerd *peer, struct peer_req *pr, struct xseg_request *xreq)
  51 {
  52         struct xseg_request *req = pr->req;
  53         if (req->state & (XS_SERVED | XS_FAILED)){
  54                 log_pr("completing", pr);
  55                 complete_forwarded(peer, pr);
  56         }
  57         else {
  58                 log_pr("forwarding", pr);
  59                 forward(peer,pr);
  60         }
  61         return 0;
  62 }
  63
  64 int mpause(struct peerd *peer)
  65 {
  66         struct xseg *xseg = peer->xseg;
  67         struct xseg_port *port = xseg_get_port(xseg, peer->portno);
  68         if (!port)
  69                 return -1;
  70
  71         xlock_acquire(&port->rq_lock, peer->portno);
  72         xlock_acquire(&port->pq_lock, peer->portno);
  73         return 0;
  74 }
  75
  76 int munpause(struct peerd *peer)
  77 {
  78         struct xseg *xseg = peer->xseg;
  79         struct xseg_port *port = xseg_get_port(xseg, peer->portno);
  80         if (!port)
  81                 return -1;
  82
  83         xlock_release(&port->rq_lock);
  84         xlock_release(&port->pq_lock);
  85         return 0;
  86 }
  87
  88 struct peerd *main_peer;
  89
  90 void main_loop(void)
  91 {
  92         int ret;
  93         struct peerd * peer = main_peer;
  94         char buf[INPUT_BUF_SIZE];
  95         char *nl;
  96
  97         unsigned int portno = NoPort, dstgw, srcgw;
  98
  99         for (;;){
 100                 printf("waitin next line\n");
 101                 if (fgets(buf, INPUT_BUF_SIZE, stdin)) {
 102                         nl = strchr(buf, '\n');
 103                         if (nl)
 104                                 *nl = 0;
 105                         buf[INPUT_BUF_SIZE -1] = 0;
 106                         printf("got line input\n");
 107                         ret = sscanf(buf, "set srcgw %u %u", &portno, &srcgw);
 108                         if (ret == 2){
 109                                 printf("found setsrcgw\n");
 110                                 xseg_set_srcgw(peer->xseg, (uint32_t) portno, (uint32_t) srcgw);
 111                                 continue;
 112                         };
 113                         ret = sscanf(buf, "set dstgw %u %u", &portno, &dstgw);
 114                         if (ret == 2){
 115                                 printf("found set dstgw\n");
 116                                 xseg_set_dstgw(peer->xseg, (uint32_t) portno, (uint32_t) dstgw);
 117                                 continue;
 118                         };
 119                         ret = sscanf(buf, "getandset srcgw %u %u", &portno, &srcgw);
 120                         if (ret == 2){
 121                                 printf("found getand set srcgw\n");
 122                                 xseg_getandset_srcgw(peer->xseg, (uint32_t) portno, (uint32_t) srcgw);
 123                                 continue;
 124                         };
 125                         ret = sscanf(buf, "getandset dstgw %u %u", &portno, &dstgw);
 126                         if (ret == 2){
 127                                 printf("found getandset dstgw\n");
 128                                 xseg_getandset_dstgw(peer->xseg, (uint32_t) portno, (uint32_t) dstgw);
 129                                 continue;
 130                         };
 131                         ret = sscanf(buf, "pause %u", &portno);
 132                         if (ret == 1){
 133                                 printf("found pause\n");
 134                                 mpause(peer);
 135                                 continue;
 136                         };
 137                         ret = sscanf(buf, "unpause %u", &portno);
 138                         if (ret == 1){
 139                                 printf("found unpause\n");
 140                                 munpause(peer);
 141                                 continue;
 142                         };
 143                 }
 144                 else
 145                         exit(0);
 146         }
 147 }
 148
 149 int custom_peer_init(struct peerd *peer, int argc, char *argv[])
 150 {
 151         int i;
 152         struct monitor_io *mio;
 153         struct monitord *monitor;
 154
 155         monitor = malloc(sizeof(struct monitord));
 156         if (!monitor)
 157                 return -1;
 158         peer->priv = monitor;
 159         monitor->mon_portno = NoPort;
 160
 161
 162         for (i = 0; i < peer->nr_ops; i++) {
 163                 mio = malloc(sizeof(struct monitor_io));
 164                 if (!mio)
 165                         return -1;
 166                 peer->peer_reqs[i].priv = mio;
 167                 mio->src_portno = NoPort;
 168         }
 169
 170         for (i = 1; i < argc; i++) {
 171                 if (!strcmp(argv[i], "-mp") && (i + 1 < argc)) {
 172                         monitor->mon_portno = atoi(argv[i+1]);
 173                         i+=1;
 174                         continue;
 175                 }
 176         }
 177         main_peer = peer;
 178
 179         peer->interactive_func = main_loop;
 180
 181         return 0;
 182 }