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 <peer.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                 enum dispatch_reason reason)
  52 {
  53         struct xseg_request *req = pr->req;
  54         if (req->state & (XS_SERVED | XS_FAILED)){
  55                 log_pr("completing", pr);
  56                 complete_forwarded(peer, pr);
  57         }
  58         else {
  59                 log_pr("forwarding", pr);
  60                 forward(peer,pr);
  61         }
  62         return 0;
  63 }
  64
  65 int mpause(struct peerd *peer)
  66 {
  67         struct xseg *xseg = peer->xseg;
  68         struct xseg_port *port = xseg_get_port(xseg, peer->portno_start);
  69         if (!port)
  70                 return -1;
  71
  72         xlock_acquire(&port->rq_lock, peer->portno_start);
  73         xlock_acquire(&port->pq_lock, peer->portno_start);
  74         return 0;
  75 }
  76
  77 int munpause(struct peerd *peer)
  78 {
  79         struct xseg *xseg = peer->xseg;
  80         struct xseg_port *port = xseg_get_port(xseg, peer->portno_start);
  81         if (!port)
  82                 return -1;
  83
  84         xlock_release(&port->rq_lock);
  85         xlock_release(&port->pq_lock);
  86         return 0;
  87 }
  88
  89 struct peerd *main_peer;
  90
  91 void main_loop(void)
  92 {
  93         int ret;
  94         struct peerd * peer = main_peer;
  95         char buf[INPUT_BUF_SIZE];
  96         char *nl;
  97
  98         unsigned int portno = NoPort, dstgw, srcgw;
  99
 100         for (;;){
 101                 printf("waitin next line\n");
 102                 if (fgets(buf, INPUT_BUF_SIZE, stdin)) {
 103                         nl = strchr(buf, '\n');
 104                         if (nl)
 105                                 *nl = 0;
 106                         buf[INPUT_BUF_SIZE -1] = 0;
 107                         printf("got line input\n");
 108                         ret = sscanf(buf, "set srcgw %u %u", &portno, &srcgw);
 109                         if (ret == 2){
 110                                 printf("found setsrcgw\n");
 111                                 xseg_set_srcgw(peer->xseg, (uint32_t) portno, (uint32_t) srcgw);
 112                                 continue;
 113                         };
 114                         ret = sscanf(buf, "set dstgw %u %u", &portno, &dstgw);
 115                         if (ret == 2){
 116                                 printf("found set dstgw\n");
 117                                 xseg_set_dstgw(peer->xseg, (uint32_t) portno, (uint32_t) dstgw);
 118                                 continue;
 119                         };
 120                         ret = sscanf(buf, "getandset srcgw %u %u", &portno, &srcgw);
 121                         if (ret == 2){
 122                                 printf("found getand set srcgw\n");
 123                                 xseg_getandset_srcgw(peer->xseg, (uint32_t) portno, (uint32_t) srcgw);
 124                                 continue;
 125                         };
 126                         ret = sscanf(buf, "getandset dstgw %u %u", &portno, &dstgw);
 127                         if (ret == 2){
 128                                 printf("found getandset dstgw\n");
 129                                 xseg_getandset_dstgw(peer->xseg, (uint32_t) portno, (uint32_t) dstgw);
 130                                 continue;
 131                         };
 132                         ret = sscanf(buf, "pause %u", &portno);
 133                         if (ret == 1){
 134                                 printf("found pause\n");
 135                                 mpause(peer);
 136                                 continue;
 137                         };
 138                         ret = sscanf(buf, "unpause %u", &portno);
 139                         if (ret == 1){
 140                                 printf("found unpause\n");
 141                                 munpause(peer);
 142                                 continue;
 143                         };
 144                 }
 145                 else
 146                         exit(0);
 147         }
 148 }
 149
 150 int custom_peer_init(struct peerd *peer, int argc, char *argv[])
 151 {
 152         int i;
 153         struct monitor_io *mio;
 154         struct monitord *monitor;
 155
 156         monitor = malloc(sizeof(struct monitord));
 157         if (!monitor)
 158                 return -1;
 159         peer->priv = monitor;
 160         monitor->mon_portno = NoPort;
 161
 162
 163         for (i = 0; i < peer->nr_ops; i++) {
 164                 mio = malloc(sizeof(struct monitor_io));
 165                 if (!mio)
 166                         return -1;
 167                 peer->peer_reqs[i].priv = mio;
 168                 mio->src_portno = NoPort;
 169         }
 170
 171         for (i = 1; i < argc; i++) {
 172                 if (!strcmp(argv[i], "-mp") && (i + 1 < argc)) {
 173                         monitor->mon_portno = atoi(argv[i+1]);
 174                         i+=1;
 175                         continue;
 176                 }
 177         }
 178         main_peer = peer;
 179
 180         peer->interactive_func = main_loop;
 181
 182         return 0;
 183 }