src_frescan/frescan_queues.c

   1 /*!
   2  * @file frescan_queues.c
   3  *
   4  * @brief FRESCAN queues to manage the packets by prio and servers
   5  *
   6  * @version 0.01
   7  *
   8  * @date 27-Feb-2008
   9  *
  10  * @author
  11  *      Daniel Sangorrin
  12  *
  13  * @comments
  14  *
  15  * This file contains the FRESCAN queues where frescan packets are stored and
  16  * managed.
  17  *
  18  * TODO: disable interrupts for mutual exclusion!!!
  19  *
  20  * @license
  21  *
  22 //----------------------------------------------------------------------
  23 //  Copyright (C) 2006 - 2009 by the FRESCOR consortium:
  24 //
  25 //    Universidad de Cantabria,              SPAIN
  26 //    University of York,                    UK
  27 //    Scuola Superiore Sant'Anna,            ITALY
  28 //    Kaiserslautern University,             GERMANY
  29 //    Univ. Politecnica  Valencia,           SPAIN
  30 //    Czech Technical University in Prague,  CZECH REPUBLIC
  31 //    ENEA                                   SWEDEN
  32 //    Thales Communication S.A.              FRANCE
  33 //    Visual Tools S.A.                      SPAIN
  34 //    Rapita Systems Ltd                     UK
  35 //    Evidence                               ITALY
  36 //
  37 //    See http://www.frescor.org
  38 //
  39 //        The FRESCOR project (FP6/2005/IST/5-034026) is funded
  40 //        in part by the European Union Sixth Framework Programme
  41 //        The European Union is not liable of any use that may be
  42 //        made of this code.
  43 //
  44 //
  45 //  based on previous work (FSF) done in the FIRST project
  46 //
  47 //   Copyright (C) 2005  Mälardalen University, SWEDEN
  48 //                       Scuola Superiore S.Anna, ITALY
  49 //                       Universidad de Cantabria, SPAIN
  50 //                       University of York, UK
  51 //
  52 // This file is part of FNA (Frescor Network Adaptation)
  53 //
  54 // FNA is free software; you can redistribute it and/or modify it
  55 // under terms of the GNU General Public License as published by the
  56 // Free Software Foundation; either version 2, or (at your option) any
  57 // later version.  FNA is distributed in the hope that it will be
  58 // useful, but WITHOUT ANY WARRANTY; without even the implied warranty
  59 // of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
  60 // General Public License for more details. You should have received a
  61 // copy of the GNU General Public License along with FNA; see file
  62 // COPYING. If not, write to the Free Software Foundation, 675 Mass Ave,
  63 // Cambridge, MA 02139, USA.
  64 //
  65 // As a special exception, including FNA header files in a file,
  66 // instantiating FNA generics or templates, or linking other files
  67 // with FNA objects to produce an executable application, does not
  68 // by itself cause the resulting executable application to be covered
  69 // by the GNU General Public License. This exception does not
  70 // however invalidate any other reasons why the executable file might be
  71 // covered by the GNU Public License.
  72 // -----------------------------------------------------------------------
  73  *
  74  */
  75
  76 #include <stdlib.h>
  77 #include <time.h>
  78
  79 #include "frescan_queues.h"
  80 #include "frescan_packets.h"
  81 #include "frescan_debug.h"
  82 #include "frescan_id.h"
  83 #include "frescan_data.h"
  84
  85 /**
  86  * frescan_pqueue_create() - creates a priority queue
  87  */
  88
  89 static inline frescan_prio_queue_t *frescan_pqueue_create(uint32_t max_prio,
  90                                                           frescan_network_t net)
  91 {
  92         int ret, prio;
  93         frescan_prio_queue_t *pq; // priority queue
  94
  95         pq = (frescan_prio_queue_t *)malloc(sizeof(frescan_prio_queue_t));
  96         if (pq == NULL) {
  97                 FRESCAN_ERROR("could not allocate memory for prio queue\n");
  98                 return NULL;
  99         }
 100
 101         pq->max_prio = max_prio;
 102         pq->net      = net;
 103
 104         ret = sem_init (&pq->sem, 0, 0);
 105         if (ret != 0) {
 106                 FRESCAN_ERROR("could not init the semaphore\n");
 107                 free(pq);
 108                 return NULL;
 109         }
 110
 111         pq->fifo_queues = (frescan_packet_t *)
 112                         malloc(max_prio * sizeof(frescan_packet_t));
 113
 114         for(prio=0; prio<max_prio; prio++) {
 115                 INIT_LIST_HEAD(&(pq->fifo_queues[prio].fifo_list));
 116         }
 117
 118         return pq;
 119 }
 120
 121 /**
 122  * frescan_queues_init() - initialize the queues
 123  *
 124  * 1.- create the transmission fixed priority queue
 125  * 2.- create the rx channels and its associated priority queues
 126  *
 127  * TODO: when error free memory
 128  */
 129
 130 int frescan_queues_init(frescan_queues_t *queues,
 131                         frescan_init_params_t *params)
 132 {
 133         int i;
 134         uint32_t max_prio;
 135
 136         // create transmission fixed priority queue
 137         queues->tx_fp_queue = frescan_pqueue_create(params->tx_fp_max_prio,
 138                                                     params->net);
 139
 140         if (queues->tx_fp_queue == NULL)  {
 141                 FRESCAN_ERROR("could not allocate memory for tx fp queue\n");
 142                 return -1;
 143         }
 144
 145         // create receiving channels
 146         queues->rx_channel_queues = (frescan_prio_queue_t **)
 147                 malloc(params->rx_num_of_channels *
 148                        sizeof(frescan_prio_queue_t *));
 149
 150         if (queues->rx_channel_queues == NULL) {
 151                 FRESCAN_ERROR
 152                         ("could not allocate memory for receiving channels\n");
 153                 return -1;
 154         }
 155
 156         queues->num_rx_channels = params->rx_num_of_channels;
 157
 158         // create a priority queue for each channel
 159         for(i=0; i<params->rx_num_of_channels; i++) {
 160
 161                 if (params->rx_channel_max_prio == NULL) {
 162                         max_prio = params->tx_fp_max_prio;
 163                 } else {
 164                         max_prio = params->rx_channel_max_prio[i];
 165                 }
 166
 167                 queues->rx_channel_queues[i] = frescan_pqueue_create
 168                                                        (max_prio, params->net);
 169
 170                 if (queues->rx_channel_queues[i] == NULL)  {
 171                         FRESCAN_ERROR
 172                                ("could not allocate memory for rx pq %d\n", i);
 173                         return -1;
 174                 }
 175         }
 176
 177         return 0;
 178 }
 179
 180 /**
 181  * frescan_pqueue_enqueue() - enqueue a packet
 182  *
 183  * check the packet flags and enqueue the packet in the appropiate queue
 184  */
 185
 186 int frescan_pqueue_enqueue(frescan_prio_queue_t *pqueue,
 187                            frescan_packet_t *packet,
 188                            frescan_prio_t prio)
 189 {
 190         int ret;
 191
 192         if (prio >= pqueue->max_prio) {
 193                 FRESCAN_ERROR("priority of the packet is too high\n");
 194                 return -1;
 195         }
 196
 197         DEBUG(FRESCAN_QUEUES_ENABLE_DEBUG,
 198               "enqueue packet with prio %u, pending %u\n",
 199               prio, packet->buffer_pending_bytes);
 200
 201         list_add_tail(&(packet->fifo_list),
 202                       &(pqueue->fifo_queues[prio].fifo_list));
 203
 204         ret = sem_post(&pqueue->sem);
 205         if (ret != 0) return ret;
 206
 207         return 0;
 208 }
 209
 210 /**
 211  * frescan_pqueue_requeue() - requeue a packet
 212  */
 213
 214 int frescan_pqueue_requeue(frescan_prio_queue_t *pqueue,
 215                            frescan_packet_t *packet,
 216                            frescan_prio_t prio)
 217 {
 218         int ret;
 219
 220         if (prio >= pqueue->max_prio) {
 221                 FRESCAN_ERROR("priority of the packet is too high\n");
 222                 return -1;
 223         }
 224
 225         list_add(&packet->fifo_list, &(pqueue->fifo_queues[prio].fifo_list));
 226
 227         ret = sem_post(&pqueue->sem);
 228         if (ret != 0) return ret;
 229
 230         return 0;
 231 }
 232
 233 /**
 234  * frescan_pqueue_dequeue() - dequeue the packet with highest priority
 235  */
 236
 237 int frescan_pqueue_dequeue(frescan_prio_queue_t *pqueue,
 238                            frescan_packet_t **packet,
 239                            frescan_prio_t *packet_prio,
 240                            bool blocking)
 241 {
 242         int prio;
 243         int ret;
 244         frescan_packet_t *tmp = NULL;
 245         struct list_head *pos;
 246
 247         *packet = NULL;
 248
 249         if (blocking) {
 250                 DEBUG(FRESCAN_QUEUES_ENABLE_DEBUG, "calling sem_wait\n");
 251                 sem_wait(&pqueue->sem);
 252         } else {
 253                 ret = sem_trywait (&pqueue->sem);
 254                 if (ret != 0) {
 255                         DEBUG(FRESCAN_QUEUES_ENABLE_DEBUG,
 256                               "sem_trywait was locked (no packets)\n");
 257                         return 0;
 258                 }
 259         }
 260
 261         DEBUG(FRESCAN_QUEUES_ENABLE_DEBUG,
 262               "check priority fifo queues (max_prio=%u)\n", pqueue->max_prio);
 263
 264         // NOTE: we only acquire the lock if we block because non-blocking
 265         // calls are made from a context with no interrupts (when updating
 266         // the buffer at 'frescan_hw_buffer_update' which is always called
 267         // with interrupts disabled)
 268         if (blocking) FRESCAN_ACQUIRE_LOCK(&frescan_data[pqueue->net].lock);
 269
 270         for(prio=pqueue->max_prio-1; prio >= 0; prio--) {
 271                 if (!list_empty(&pqueue->fifo_queues[prio].fifo_list)) {
 272                         list_for_each(pos,
 273                                       &pqueue->fifo_queues[prio].fifo_list) {
 274                                 tmp = list_entry(pos, frescan_packet_t,
 275                                                  fifo_list);
 276                                 break;
 277                         }
 278                         *packet = tmp;
 279                         list_del(&tmp->fifo_list);
 280                         break;
 281                 }
 282         }
 283
 284         if (blocking) FRESCAN_RELEASE_LOCK(&frescan_data[pqueue->net].lock);
 285
 286         DEBUG(FRESCAN_QUEUES_ENABLE_DEBUG, "dequeued prio %u\n", prio);
 287         *packet_prio = prio;
 288
 289         return 0;
 290 }
 291
 292 /**
 293  * frescan_pqueue_get_highest_prio() - returns the packet with highest prio
 294  * but not extracting it from the queue.
 295  */
 296
 297 int frescan_pqueue_get_highest_prio(frescan_prio_queue_t *pqueue,
 298                                     frescan_packet_t **packet,
 299                                     frescan_prio_t *packet_prio)
 300 {
 301         int prio;
 302         frescan_packet_t *tmp = NULL;
 303         struct list_head *pos;
 304
 305         *packet = NULL;
 306
 307         for(prio=pqueue->max_prio-1; prio >= 0; prio--) {
 308                 if (!list_empty(&pqueue->fifo_queues[prio].fifo_list)) {
 309                         list_for_each(pos,
 310                                       &pqueue->fifo_queues[prio].fifo_list) {
 311                                 tmp = list_entry(pos, frescan_packet_t,
 312                                                  fifo_list);
 313                                 break;
 314                         }
 315                         *packet = tmp;
 316                         break;
 317                 }
 318         }
 319
 320         DEBUG(FRESCAN_QUEUES_ENABLE_DEBUG, "highest prio %u\n", prio);
 321         *packet_prio = prio;
 322
 323         return 0;
 324 }
 325
 326
 327 /**
 328  * frescan_servers_enqueue() - enqueue a packet through a server
 329  *
 330  * @net: the network instance
 331  * @id: the identificator for the server
 332  * @packet: the packet being enqueued
 333  *
 334  */
 335
 336 int frescan_servers_enqueue(frescan_network_t net,
 337                             frescan_ss_t id,
 338                             frescan_packet_t *packet)
 339 {
 340         frescan_ss_data_t *server = &frescan_data[net].ss_data[id];
 341
 342         clock_gettime (CLOCK_MONOTONIC, &packet->timestamp);
 343
 344         DEBUG(FRESCAN_SERVERS_ENABLE_DEBUG, "id:%u timestamp:(%d, %d)\n",
 345               id, packet->timestamp.tv_sec, packet->timestamp.tv_nsec);
 346
 347         // add the packet to the server fifo list
 348         list_add_tail(&packet->fifo_list, &server->packet_list.fifo_list);
 349
 350         // if the server was inactive (no packets) put it in the active list
 351         if (server->pending_packets == 0) {
 352                 DEBUG(FRESCAN_SERVERS_ENABLE_DEBUG,
 353                       "ss becomes active act_time=timestamp\n");
 354                 list_add_tail(&server->servers_list,
 355                                &frescan_data[net].ss_active_head.servers_list);
 356                 server->act_time = packet->timestamp;
 357         }
 358
 359         server->pending_packets++;
 360         return 0;
 361 }
 362
 363 /**
 364  * frescan_servers_requeue() - requeue a packet through a server
 365  *
 366  * @net: the network instance
 367  * @id: the identificator for the server
 368  * @packet: the packet being requeued
 369  *
 370  */
 371
 372 int frescan_servers_requeue(frescan_network_t net,
 373                             frescan_ss_t id,
 374                             frescan_packet_t *packet)
 375 {
 376         DEBUG(FRESCAN_SERVERS_ENABLE_DEBUG,
 377               "requeue packet for id:%u\n", id);
 378
 379         // add the packet to the server fifo list
 380         list_add(&packet->fifo_list,
 381                   &frescan_data[net].ss_data[id].packet_list.fifo_list);
 382
 383         // if the server was inactive (no packets to send) put it active
 384         // (in the active list)
 385         if (frescan_data[net].ss_data[id].pending_packets == 0) {
 386                 DEBUG(FRESCAN_SERVERS_ENABLE_DEBUG,
 387                       "server was inactive, put in the active list\n");
 388                 list_add(&frescan_data[net].ss_data[id].servers_list,
 389                           &frescan_data[net].ss_active_head.servers_list);
 390         }
 391
 392         frescan_data[net].ss_data[id].pending_packets++;
 393         return 0;
 394 }
 395
 396 /**
 397  * frescan_servers_dequeue() - dequeue a packet from a server
 398  *
 399  * @net: the network instance
 400  * @id: the identificator for the server
 401  * @packet: the packet dequeued
 402  * @packet_prio: the priority current of the server
 403  *
 404  */
 405
 406 int frescan_servers_dequeue(frescan_network_t net,
 407                             frescan_ss_t id,
 408                             frescan_packet_t **packet,
 409                             frescan_prio_t *packet_prio)
 410 {
 411         struct list_head *pos;
 412         frescan_ss_data_t *server;
 413
 414         server = &frescan_data[net].ss_data[id];
 415
 416         if (list_empty(&server->packet_list.fifo_list)) {
 417                 FRESCAN_ERROR("no packet in server %d fifo list\n", id);
 418                 return -1;
 419         }
 420
 421         list_for_each(pos, &server->packet_list.fifo_list) {
 422                 *packet = list_entry(pos, frescan_packet_t, fifo_list);
 423                 break;
 424         }
 425
 426         list_del(&((*packet)->fifo_list));
 427         *packet_prio = server->current_priority;
 428         server->pending_packets--;
 429
 430         if (server->pending_packets == 0) {
 431                 DEBUG(FRESCAN_SERVERS_ENABLE_DEBUG,
 432                       "no more packets, delete from active list\n");
 433                 list_del(&server->servers_list);
 434         }
 435
 436         DEBUG(FRESCAN_SERVERS_ENABLE_DEBUG,
 437               "dequeued packet server:%u cur_prio:%u pending:%u\n",
 438               id, *packet_prio, server->pending_packets);
 439
 440         return 0;
 441 }