zephyr/zhm/queue.c

   1 /* This file is part of the Project Athena Zephyr Notification System.
   2  * It contains the hostmanager queueing routines.
   3  *
   4  *      Created by:     David C. Jedlinsky
   5  *
   6  *      $Id$
   7  *
   8  *      Copyright (c) 1987 by the Massachusetts Institute of Technology.
   9  *      For copying and distribution information, see the file
  10  *      "mit-copyright.h".
  11  */
  12
  13 #include "zhm.h"
  14
  15 #ifndef lint
  16 #ifndef SABER
  17 static char rcsid_queue_c[] = "$Id$";
  18 #endif /* SABER */
  19 #endif /* lint */
  20
  21 typedef struct _Queue {
  22     Timer *timer;
  23     int retries;
  24     ZNotice_t notice;
  25     caddr_t packet;
  26     struct sockaddr_in reply;
  27     struct _Queue *next, **prev_p;
  28 } Queue;
  29
  30 static Queue *hm_queue;
  31 static int retransmits_enabled = 0;
  32
  33 static Queue *find_notice_in_queue(ZNotice_t *notice);
  34 static Code_t dump_queue(void);
  35 static void queue_timeout(void *arg);
  36
  37 int rexmit_times[] = { 2, 2, 4, 4, 8, -1 };
  38
  39 #ifdef DEBUG
  40 Code_t dump_queue(void);
  41 #endif
  42
  43 void
  44 init_queue(void)
  45 {
  46     Queue *q;
  47
  48     while (hm_queue) {
  49         q = hm_queue;
  50         if (q->timer)
  51             timer_reset(q->timer);
  52         free(q->packet);
  53         hm_queue = q->next;
  54         free(q);
  55     }
  56
  57     DPR("Queue initialized and flushed.\n");
  58 }
  59
  60 Code_t
  61 add_notice_to_queue(ZNotice_t *notice,
  62                     char *packet,
  63                     struct sockaddr_in *repl,
  64                     int len)
  65 {
  66     Queue *entry;
  67
  68     DPR("Adding notice to queue...\n");
  69     if (!find_notice_in_queue(notice)) {
  70         entry = (Queue *) malloc(sizeof(Queue));
  71         if (entry == NULL)
  72             return(ZERR_NONOTICE);
  73         entry->retries = 0;
  74         entry->packet = (char *) malloc(Z_MAXPKTLEN);
  75         if (entry->packet == NULL) {
  76             free(entry);
  77             return(ZERR_NONOTICE);
  78         }
  79         memcpy(entry->packet, packet, Z_MAXPKTLEN);
  80         if (ZParseNotice(entry->packet, len, &entry->notice) != ZERR_NONE) {
  81             syslog(LOG_ERR, "ZParseNotice failed, but succeeded before");
  82             free(entry->packet);
  83         } else {
  84             entry->reply = *repl;
  85             LIST_INSERT(&hm_queue, entry);
  86         }
  87         entry->timer = (retransmits_enabled) ?
  88             timer_set_rel(rexmit_times[0], queue_timeout, entry) : NULL;
  89     }
  90     return(ZERR_NONE);
  91 }
  92
  93 Code_t
  94 remove_notice_from_queue(ZNotice_t *notice,
  95                          ZNotice_Kind_t *kind,
  96                          struct sockaddr_in *repl)
  97 {
  98     Queue *entry;
  99
 100     DPR("Removing notice from queue...\n");
 101     entry = find_notice_in_queue(notice);
 102     if (entry == NULL)
 103         return(ZERR_NONOTICE);
 104
 105     *kind = entry->notice.z_kind;
 106     *repl = entry->reply;
 107     if (entry->timer)
 108         timer_reset(entry->timer);
 109     free(entry->packet);
 110     LIST_DELETE(entry);
 111 #ifdef DEBUG
 112     dump_queue();
 113 #endif /* DEBUG */
 114     free(entry);
 115     return(ZERR_NONE);
 116 }
 117
 118 /* We have a server; transmit all of our packets. */
 119 void
 120 retransmit_queue(struct sockaddr_in *sin)
 121 {
 122     Queue *entry;
 123     Code_t ret;
 124
 125     DPR("Retransmitting queue to new server...\n");
 126     ret = ZSetDestAddr(sin);
 127     if (ret != ZERR_NONE) {
 128         Zperr (ret);
 129         com_err("queue", ret, "setting destination");
 130     }
 131     for (entry = hm_queue; entry; entry = entry->next) {
 132         DPR("notice:\n");
 133         DPR2("\tz_kind: %d\n", entry->notice.z_kind);
 134         DPR2("\tz_port: %u\n", ntohs(entry->notice.z_port));
 135         DPR2("\tz_class: %s\n", entry->notice.z_class);
 136         DPR2("\tz_clss_inst: %s\n", entry->notice.z_class_inst);
 137         DPR2("\tz_opcode: %s\n", entry->notice.z_opcode);
 138         DPR2("\tz_sender: %s\n", entry->notice.z_sender);
 139         DPR2("\tz_recip: %s\n", entry->notice.z_recipient);
 140         ret = send_outgoing(&entry->notice);
 141         if (ret != ZERR_NONE) {
 142             Zperr(ret);
 143             com_err("queue", ret, "sending raw notice");
 144         }
 145         entry->timer = timer_set_rel(rexmit_times[0], queue_timeout, entry);
 146         entry->retries = 0;
 147     }
 148     retransmits_enabled = 1;
 149 }
 150
 151 /* We lost our server; nuke all of our timers. */
 152 void
 153 disable_queue_retransmits(void)
 154 {
 155     Queue *entry;
 156
 157     for (entry = hm_queue; entry; entry = entry->next) {
 158         if (entry->timer)
 159             timer_reset(entry->timer);
 160         entry->timer = NULL;
 161     }
 162     retransmits_enabled = 0;
 163 }
 164
 165 #ifdef DEBUG
 166 static Code_t
 167 dump_queue(void)
 168 {
 169     Queue *entry;
 170     caddr_t mp;
 171     int ml;
 172
 173     DPR("Dumping queue...\n");
 174     if (!hm_queue) {
 175         printf("Queue is empty.\n");
 176         return;
 177     }
 178
 179     for (entry = hm_queue; entry; entry = entry->next) {
 180         printf("notice:\n");
 181         printf("\tz_kind: %d\n", entry->notice.z_kind);
 182         printf("\tz_port: %u\n", ntohs(entry->notice.z_port));
 183         printf("\tz_class: %s\n", entry->notice.z_class);
 184         printf("\tz_clss_inst: %s\n", entry->notice.z_class_inst);
 185         printf("\tz_opcode: %s\n", entry->notice.z_opcode);
 186         printf("\tz_sender: %s\n", entry->notice.z_sender);
 187         printf("\tz_recip: %s\n", entry->notice.z_recipient);
 188         printf("\tMessage:\n");
 189         mp = entry->notice.z_message;
 190         for (ml = strlen(mp) + 1; ml <= entry->notice.z_message_len; ml++) {
 191             printf("\t%s\n", mp);
 192             mp += strlen(mp)+1;
 193             ml += strlen(mp);
 194         }
 195     }
 196 }
 197 #endif /* DEBUG */
 198
 199 int
 200 queue_len(void)
 201 {
 202     int length = 0;
 203     Queue *entry;
 204
 205     for (entry = hm_queue; entry; entry = entry->next)
 206         length++;
 207     return length;
 208 }
 209
 210 static Queue *
 211 find_notice_in_queue(ZNotice_t *notice)
 212 {
 213     Queue *entry;
 214
 215     for (entry = hm_queue; entry; entry = entry->next) {
 216         if (ZCompareUID(&entry->notice.z_uid, &notice->z_uid))
 217             return entry;
 218     }
 219     return NULL;
 220 }
 221
 222 static void
 223 queue_timeout(void *arg)
 224 {
 225     Queue *entry = (Queue *) arg;
 226     Code_t ret;
 227
 228     entry->timer = NULL;
 229     ret = ZSetDestAddr(&serv_sin);
 230     if (ret != ZERR_NONE) {
 231         Zperr(ret);
 232         com_err("queue", ret, "setting destination");
 233     }
 234     entry->retries++;
 235     if (rexmit_times[entry->retries] == -1) {
 236         new_server(NULL);
 237         return;
 238     }
 239     DPR("Resending notice:\n");
 240     DPR2("\tz_kind: %d\n", entry->notice.z_kind);
 241     DPR2("\tz_port: %u\n", ntohs(entry->notice.z_port));
 242     DPR2("\tz_class: %s\n", entry->notice.z_class);
 243     DPR2("\tz_clss_inst: %s\n", entry->notice.z_class_inst);
 244     DPR2("\tz_opcode: %s\n", entry->notice.z_opcode);
 245     DPR2("\tz_sender: %s\n", entry->notice.z_sender);
 246     DPR2("\tz_recip: %s\n", entry->notice.z_recipient);
 247     ret = send_outgoing(&entry->notice);
 248     if (ret != ZERR_NONE) {
 249         Zperr(ret);
 250         com_err("queue", ret, "sending raw notice");
 251     }
 252     entry->timer = timer_set_rel(rexmit_times[entry->retries], queue_timeout,
 253                                  entry);
 254 }
 255