src/xt_engine.c

   1 /*
   2  * Lightweight Autonomic Network Architecture
   3  *
   4  * LANA packet processing engines. Incoming packtes are scheduled onto one
   5  * of the CPU-affine engines and processed on the Functional Block stack.
   6  * There are two queues where packets can be added, one from PHY direction
   7  * for incoming packets (ingress) and one from the socket handler direction
   8  * for outgoing packets (egress). Support for NUMA-affinity added.
   9  *
  10  * Copyright 2011 Daniel Borkmann <dborkma@tik.ee.ethz.ch>,
  11  * Swiss federal institute of technology (ETH Zurich)
  12  * Subject to the GPL.
  13  */
  14
  15 #include <linux/cpu.h>
  16 #include <linux/kernel.h>
  17 #include <linux/slab.h>
  18 #include <linux/skbuff.h>
  19 #include <linux/wait.h>
  20 #include <linux/kthread.h>
  21 #include <linux/proc_fs.h>
  22 #include <linux/u64_stats_sync.h>
  23 #include <linux/prefetch.h>
  24 #include <linux/sched.h>
  25
  26 #include "xt_engine.h"
  27 #include "xt_skb.h"
  28 #include "xt_fblock.h"
  29
  30 struct worker_engine __percpu *engines;
  31 EXPORT_SYMBOL_GPL(engines);
  32 extern struct proc_dir_entry *lana_proc_dir;
  33
  34 void cleanup_worker_engines(void);
  35
  36 static inline struct ppe_queue *first_ppe_queue(struct worker_engine *ppe)
  37 {
  38         return ppe->inqs.head;
  39 }
  40
  41 static inline struct ppe_queue *next_filled_ppe_queue(struct ppe_queue *ppeq)
  42 {
  43         do {
  44                 ppeq = ppeq->next;
  45                 prefetch(ppeq->next);
  46         } while (skb_queue_empty(&ppeq->queue));
  47
  48         return ppeq;
  49 }
  50
  51 static inline int ppe_queues_have_load(struct worker_engine *ppe)
  52 {
  53         return atomic64_read(&ppe->load) != 0;
  54 }
  55
  56 static inline void ppe_queues_reduce_load(struct worker_engine *ppe)
  57 {
  58         atomic64_dec(&ppe->load);
  59 }
  60
  61 static int process_packet(struct sk_buff *skb, enum path_type dir)
  62 {
  63         int ret = PPE_DROPPED;
  64         idp_t cont;
  65         struct fblock *fb;
  66
  67         while ((cont = read_next_idp_from_skb(skb))) {
  68                 fb = search_fblock(cont);
  69                 if (unlikely(!fb)) {
  70                         ret = PPE_ERROR;
  71                         break;
  72                 }
  73                 ret = fb->ops->netfb_rx(fb, skb, &dir);
  74                 put_fblock(fb);
  75                 if (ret == PPE_DROPPED)
  76                         break;
  77         }
  78
  79         return ret;
  80 }
  81
  82 static int engine_thread(void *arg)
  83 {
  84         int ret;
  85         struct sk_buff *skb;
  86         struct ppe_queue *ppeq;
  87         struct worker_engine *ppe = per_cpu_ptr(engines,
  88                                                 smp_processor_id());
  89
  90         if (ppe->cpu != smp_processor_id())
  91                 panic("[lana] Engine scheduled on wrong CPU!\n");
  92         printk(KERN_INFO "[lana] Packet Processing Engine running "
  93                "on CPU%u!\n", smp_processor_id());
  94
  95         ppeq = first_ppe_queue(ppe);
  96         while (1) {
  97                 wait_event_interruptible(ppe->wait_queue,
  98                                          (kthread_should_stop() ||
  99                                           ppe_queues_have_load(ppe)));
 100                 if (unlikely(kthread_should_stop()))
 101                         break;
 102
 103                 ppeq = next_filled_ppe_queue(ppeq);
 104                 ppe_queues_reduce_load(ppe);
 105                 skb = skb_dequeue(&ppeq->queue);
 106                 ret = process_packet(skb, ppeq->type);
 107
 108                 u64_stats_update_begin(&ppeq->stats.syncp);
 109                 ppeq->stats.packets++;
 110                 ppeq->stats.bytes += skb->len;
 111                 u64_stats_update_end(&ppeq->stats.syncp);
 112                 if (unlikely(ret == PPE_DROPPED)) {
 113                         u64_stats_update_begin(&ppeq->stats.syncp);
 114                         ppeq->stats.dropped++;
 115                         u64_stats_update_end(&ppeq->stats.syncp);
 116                 } else if (unlikely(ret == PPE_ERROR)) {
 117                         ppeq->stats.errors++;
 118                 }
 119
 120                 kfree_skb(skb);
 121         }
 122
 123         printk(KERN_INFO "[lana] Packet Processing Engine stopped "
 124                "on CPU%u!\n", smp_processor_id());
 125         return 0;
 126 }
 127
 128 static int engine_procfs_stats(char *page, char **start, off_t offset,
 129                                int count, int *eof, void *data)
 130 {
 131         int i;
 132         off_t len = 0;
 133         struct worker_engine *ppe = data;
 134         unsigned int sstart;
 135
 136         len += sprintf(page + len, "engine: %p\n", ppe);
 137         len += sprintf(page + len, "cpu: %u, numa node: %d\n",
 138                        ppe->cpu, cpu_to_node(ppe->cpu));
 139         len += sprintf(page + len, "load: %lld\n",
 140                        atomic64_read(&ppe->load));
 141         for (i = 0; i < NUM_TYPES; ++i) {
 142                 do {
 143                         sstart = u64_stats_fetch_begin(&ppe->inqs.ptrs[i]->stats.syncp);
 144                         len += sprintf(page + len, "queue: %p\n",
 145                                        ppe->inqs.ptrs[i]);
 146                         len += sprintf(page + len, "  type: %u\n",
 147                                        ppe->inqs.ptrs[i]->type);
 148                         len += sprintf(page + len, "  packets: %llu\n",
 149                                        ppe->inqs.ptrs[i]->stats.packets);
 150                         len += sprintf(page + len, "  bytes: %llu\n",
 151                                        ppe->inqs.ptrs[i]->stats.bytes);
 152                         len += sprintf(page + len, "  errors: %u\n",
 153                                        ppe->inqs.ptrs[i]->stats.errors);
 154                         len += sprintf(page + len, "  drops: %llu\n",
 155                                        ppe->inqs.ptrs[i]->stats.dropped);
 156                 } while (u64_stats_fetch_retry(&ppe->inqs.ptrs[i]->stats.syncp, sstart));
 157         }
 158         /* FIXME: fits in page? */
 159         *eof = 1;
 160         return len;
 161 }
 162
 163 static inline void add_to_ppe_squeue(struct ppe_squeue *qs,
 164                                      struct ppe_queue *q)
 165 {
 166         q->next = qs->head;
 167         qs->head = q;
 168         qs->ptrs[q->type] = q;
 169 }
 170
 171 static void finish_ppe_squeue(struct ppe_squeue *qs)
 172 {
 173         struct ppe_queue *q = qs->head;
 174         while (q->next)
 175                 q = q->next;
 176         q->next = qs->head;
 177 }
 178
 179 static int init_ppe_squeue(struct ppe_squeue *queues, unsigned int cpu)
 180 {
 181         int i;
 182         struct ppe_queue *tmp;
 183
 184         for (i = 0; i < NUM_TYPES; ++i) {
 185                 tmp = kzalloc_node(sizeof(*tmp), GFP_KERNEL,
 186                                    cpu_to_node(cpu));
 187                 if (!tmp)
 188                         return -ENOMEM;
 189                 tmp->type = (enum path_type) i;
 190                 tmp->next = NULL;
 191                 skb_queue_head_init(&tmp->queue);
 192                 add_to_ppe_squeue(queues, tmp);
 193         }
 194
 195         finish_ppe_squeue(queues);
 196         return 0;
 197 }
 198
 199 static void cleanup_ppe_squeue(struct ppe_squeue *queues)
 200 {
 201         int i;
 202
 203         for (i = 0; i < NUM_TYPES; ++i) {
 204                 if (queues->ptrs[i])
 205                         kfree(queues->ptrs[i]);
 206                 queues->ptrs[i] = NULL;
 207         }
 208         queues->head = NULL;
 209 }
 210
 211 int init_worker_engines(void)
 212 {
 213         int ret = 0;
 214         unsigned int cpu;
 215         char name[64];
 216         struct sched_param param = { .sched_priority = MAX_RT_PRIO-1 };
 217
 218         engines = alloc_percpu(struct worker_engine);
 219         if (!engines)
 220                 return -ENOMEM;
 221
 222         get_online_cpus();
 223         for_each_online_cpu(cpu) {
 224                 struct worker_engine *ppe;
 225                 ppe = per_cpu_ptr(engines, cpu);
 226                 ppe->cpu = cpu;
 227                 ppe->inqs.head = NULL;
 228                 memset(&ppe->inqs, 0, sizeof(ppe->inqs));
 229                 ret = init_ppe_squeue(&ppe->inqs, ppe->cpu);
 230                 if (ret < 0)
 231                         break;
 232                 atomic64_set(&ppe->load, 0);
 233                 memset(name, 0, sizeof(name));
 234                 snprintf(name, sizeof(name), "ppe%u", cpu);
 235                 ppe->proc = create_proc_read_entry(name, 0400, lana_proc_dir,
 236                                                    engine_procfs_stats, ppe);
 237                 if (!ppe->proc) {
 238                         ret = -ENOMEM;
 239                         break;
 240                 }
 241
 242                 init_waitqueue_head(&ppe->wait_queue);
 243                 ppe->thread = kthread_create_on_node(engine_thread, NULL,
 244                                                      cpu_to_node(cpu), name);
 245                 if (IS_ERR(ppe->thread)) {
 246                         printk(KERN_ERR "[lana] Error creationg thread on "
 247                                "node %u!\n", cpu);
 248                         ret = -EIO;
 249                         break;
 250                 }
 251
 252                 kthread_bind(ppe->thread, cpu);
 253                 sched_setscheduler(ppe->thread, SCHED_FIFO, &param);
 254                 wake_up_process(ppe->thread);
 255         }
 256         put_online_cpus();
 257
 258         if (ret < 0)
 259                 cleanup_worker_engines();
 260         return ret;
 261 }
 262 EXPORT_SYMBOL_GPL(init_worker_engines);
 263
 264 void cleanup_worker_engines(void)
 265 {
 266         unsigned int cpu;
 267         char name[64];
 268
 269         get_online_cpus();
 270         for_each_online_cpu(cpu) {
 271                 struct worker_engine *ppe;
 272                 memset(name, 0, sizeof(name));
 273                 snprintf(name, sizeof(name), "ppe%u", cpu);
 274                 ppe = per_cpu_ptr(engines, cpu);
 275                 if (!IS_ERR(ppe->thread))
 276                         kthread_stop(ppe->thread);
 277                 if (ppe->proc)
 278                         remove_proc_entry(name, lana_proc_dir);
 279                 cleanup_ppe_squeue(&ppe->inqs);
 280         }
 281         put_online_cpus();
 282         free_percpu(engines);
 283 }
 284 EXPORT_SYMBOL_GPL(cleanup_worker_engines);
 285