Index: ksyslog/trunk/ksyslog.c
===================================================================
--- ksyslog/trunk/ksyslog.c	(revision 270)
+++ ksyslog/trunk/ksyslog.c	(revision 271)
@@ -13,5 +13,4 @@
 #include <linux/fsnotify.h>
 #include <linux/proc_fs.h>
-#include <linux/u64_stats_sync.h>
 #include <linux/percpu.h>
 #include <net/udp.h>
@@ -19,15 +18,12 @@
 #include "ksyslog.h"
 
-static struct ksyslog_queue ksyslog_queue;
-static struct socket *ksyslog_rcv_sk = NULL;
-
-static struct delayed_work ksyslog_work;
-static struct workqueue_struct *ksyslog_wq = NULL;
+static struct socket *ksyslog_rcv_sk;
+static struct workqueue_struct *ksyslog_wq;
+static struct ksyslog_queue __percpu *ksyslog_queue;
 
 #ifdef CONFIG_PROC_FS
-static struct proc_dir_entry *ksyslog_procdir = NULL;
-static struct proc_dir_entry *ksyslog_proc_queue = NULL;
-static struct proc_dir_entry *ksyslog_proc_size = NULL;
-static struct proc_dir_entry *ksyslog_proc_stats = NULL;
+static struct proc_dir_entry *ksyslog_procdir;
+static struct proc_dir_entry *ksyslog_proc_size;
+static struct proc_dir_entry *ksyslog_proc_stats;
 #endif
 
@@ -36,7 +32,4 @@
 static char *ksyslog_path = "/var/log/ksyslog.log";
 static ulong ksyslog_queue_size_max = 4096;
-static ulong ksyslog_flush_interval = 45;  /* milliseconds */
-
-static DEFINE_SPINLOCK(ksyslog_write_lock);
 
 module_param(ksyslog_host, charp, 0444);
@@ -44,25 +37,36 @@
 module_param(ksyslog_path, charp, 0644);
 module_param(ksyslog_queue_size_max, ulong, 0644);
-module_param(ksyslog_flush_interval, ulong, 0644);
-
-static int
-ksyslog_queue_init(struct ksyslog_queue *queue)
-{
-	memset(queue, 0, sizeof(*queue));
-	INIT_LIST_HEAD(&queue->head);
-	spin_lock_init(&queue->lock);
-	atomic64_set(&queue->size, 0);
-	queue->stats = alloc_percpu(struct ksyslog_stats);
-	if (unlikely(queue->stats == NULL))
+
+static int
+ksyslog_queue_init(void (*handler)(struct work_struct *))
+{
+	int cpu;
+	struct ksyslog_queue *q;
+
+	ksyslog_queue = alloc_percpu(struct ksyslog_queue);
+	if (unlikely(!ksyslog_queue))
 		return -ENOMEM;
+
+	for_each_possible_cpu(cpu) {
+		q = per_cpu_ptr(ksyslog_queue, cpu);
+
+		INIT_LIST_HEAD(&q->head);
+		INIT_WORK(&q->work, handler);
+		spin_lock_init(&q->lock);
+		atomic64_set(&q->size, 0);
+		ksyslog_stats_zero(&q->write_stats);
+		ksyslog_stats_zero(&q->drop_stats);
+		ksyslog_stats_zero(&q->discard_stats);
+	}
+
 	return 0;
 }
 
 static void
-ksyslog_queue_uninit(struct ksyslog_queue *queue)
-{
-	if (likely(queue->stats))
-		free_percpu(queue->stats);
-	queue->stats = NULL;
+ksyslog_queue_uninit(void)
+{
+	if (likely(ksyslog_queue))
+		free_percpu(ksyslog_queue);
+	ksyslog_queue = NULL;
 }
 
@@ -161,5 +165,5 @@
 
 	start = memchr(skb->data, '>', 5);
-	if (start == NULL)
+	if (!start)
 		return ERR_PTR(-EINVAL);
 	start++;
@@ -186,10 +190,10 @@
 
 	entry = kzalloc(sizeof(*entry), GFP_ATOMIC);
-	if (unlikely(entry == NULL))
+	if (unlikely(!entry))
 		return ERR_PTR(-ENOMEM);
 
 	length = skb->len - (start - skb->data);
 	entry->data = kzalloc(length, GFP_ATOMIC);
-	if (unlikely(entry->data == NULL)) {
+	if (unlikely(!entry->data)) {
 		kfree(entry);
 		return ERR_PTR(-ENOMEM);
@@ -264,5 +268,5 @@
 {
 	*buf = kzalloc(54 + entry->length + 2, GFP_ATOMIC);
-	if (unlikely(*buf == NULL))
+	if (unlikely(!*buf))
 		return -ENOMEM;
 
@@ -297,56 +301,39 @@
 
 static void
-ksyslog_work_register(unsigned long timer)
-{
-	queue_delayed_work(ksyslog_wq, &ksyslog_work, timer * HZ / 1000);
-}
-
-static void
-ksyslog_work_unregister(void)
-{
-	cancel_delayed_work_sync(&ksyslog_work);
-}
-
-static void
 ksyslog_work_handler(struct work_struct *work)
 {
-	struct file *file = NULL;
+	struct file *file;
 	struct ksyslog_entry *entry;
+	struct ksyslog_queue *q;
+
+	q = container_of(work, struct ksyslog_queue, work);
 
 	file = ksyslog_open(ksyslog_path);
 	if (unlikely(IS_ERR(file)))
-		goto out;
-
-	while (true) {
-		bool write_ok;
-
-		spin_lock_bh(&ksyslog_queue.lock);
-		entry = list_first_or_null_rcu(&ksyslog_queue.head,
-					       struct ksyslog_entry, list);
-		if (!entry) {
-			spin_unlock_bh(&ksyslog_queue.lock);
-			break;
-		}
-		ksyslog_entry_del(&ksyslog_queue, entry, false);
-		spin_unlock_bh(&ksyslog_queue.lock);
-
-		spin_lock(&ksyslog_write_lock);
-		write_ok = ksyslog_entry_write(file, entry);
-		spin_unlock(&ksyslog_write_lock);
-
-		if (likely(write_ok)) {
-			ksyslog_stats_add_write(&ksyslog_queue, entry->length);
-		} else {
-			ksyslog_stats_add_drop(&ksyslog_queue, entry->length);
-			ksyslog_drop_warning(entry);
-		}
-
-		call_rcu(&entry->rcu, ksyslog_entry_free);
-	}
-
+		return;
+
+	spin_lock_bh(&q->lock);
+	entry = list_first_or_null_rcu(&q->head, struct ksyslog_entry, list);
+	if (unlikely(!entry)) {
+		spin_unlock_bh(&q->lock);
+		goto out;
+	}
+	ksyslog_entry_del(q, entry, false);
+	spin_unlock_bh(&q->lock);
+
+	if (likely(ksyslog_entry_write(file, entry))) {
+		ksyslog_stats_add(&q->write_stats, entry->length);
+	} else {
+		ksyslog_stats_add(&q->drop_stats, entry->length);
+		ksyslog_drop_warning(entry);
+	}
+
+	call_rcu(&entry->rcu, ksyslog_entry_free);
+
+out:
 	ksyslog_close(file);
 
-out:
-	ksyslog_work_register(ksyslog_flush_interval);
+	if (atomic64_read(&q->size) > 0)
+		queue_work(ksyslog_wq, work);
 }
 
@@ -358,7 +345,10 @@
 	struct udphdr *udph;
 	struct ksyslog_entry *entry;
+	struct ksyslog_queue *q;
+
+	q = per_cpu_ptr(ksyslog_queue, smp_processor_id());
 
 	if (unlikely(skb_linearize(skb))) {
-		ksyslog_stats_add_drop(&ksyslog_queue, skb->len);
+		ksyslog_stats_add(&q->drop_stats, skb->len);
 		goto out;
 	}
@@ -368,5 +358,5 @@
 
 	if (unlikely(!skb_pull(skb, sizeof(*udph)))) {
-		ksyslog_stats_add_drop(&ksyslog_queue, skb->len);
+		ksyslog_stats_add(&q->drop_stats, skb->len);
 		goto out;
 	}
@@ -375,22 +365,24 @@
 	if (unlikely(IS_ERR(entry))) {
 		if (PTR_ERR(entry) == -EINVAL) {
-			ksyslog_stats_add_discard(&ksyslog_queue, skb->len);
+			ksyslog_stats_add(&q->discard_stats, skb->len);
 			goto out;
 		}
 
-		ksyslog_stats_add_drop(&ksyslog_queue, skb->len);
-		goto out;
-	}
-
-	spin_lock_bh(&ksyslog_queue.lock);
-	err = ksyslog_entry_add(&ksyslog_queue, entry);
-	spin_unlock_bh(&ksyslog_queue.lock);
+		ksyslog_stats_add(&q->drop_stats, skb->len);
+		goto out;
+	}
+
+	spin_lock_bh(&q->lock);
+	err = ksyslog_entry_add(q, entry);
+	spin_unlock_bh(&q->lock);
 
 	if (unlikely(err)) {
-		ksyslog_stats_add_drop(&ksyslog_queue, entry->length);
+		ksyslog_stats_add(&q->drop_stats, entry->length);
 		ksyslog_drop_warning(entry);
 		ksyslog_entry_free(&entry->rcu);
 		goto out;
 	}
+
+	queue_work(ksyslog_wq, &q->work);
 
 out:
@@ -400,82 +392,21 @@
 
 #ifdef CONFIG_PROC_FS
-static void *
-ksyslog_rculist_seq_start(struct seq_file *seq, loff_t *pos)
-{
-	struct list_head *lh, *head = seq->private;
-	loff_t ppos = *pos;
-
-	rcu_read_lock();
-
-	__list_for_each_rcu(lh, head)
-		if (ppos-- == 0)
-			return lh;
-
-	return NULL;
-}
-
-static void *
-ksyslog_rculist_seq_next(struct seq_file *seq, void *v, loff_t *pos)
-{
-	struct list_head *lh = rcu_dereference(((struct list_head *)v)->next);
-	++(*pos);
-	return lh == seq->private ? NULL : lh;
-}
-
-static void
-ksyslog_rculist_seq_stop(struct seq_file *seq, void *v)
-{
-	rcu_read_unlock();
-}
-
-static int
-ksyslog_queue_seq_show(struct seq_file *seq, void *v)
-{
-	const struct ksyslog_entry *entry = list_entry_rcu(v, struct ksyslog_entry, list);
-
-	seq_printf(seq, "%llu %s.%s %u.%u.%u.%u %.*s\n",
-		   timeval_to_ns(&entry->tv) / 1000 / 1000 / 1000,
-		   ksyslog_facility_str(entry->facility),
-		   ksyslog_severity_str(entry->severity),
-		   entry->saddr.addr8[0], entry->saddr.addr8[1],
-		   entry->saddr.addr8[2], entry->saddr.addr8[3],
-		   (int)entry->length, entry->data);
-
+static int
+ksyslog_size_seq_show(struct seq_file *seq, void *v)
+{
+	int cpu;
+	struct ksyslog_queue *q;
+
+	seq_puts(seq, "{\n");
+
+	for_each_possible_cpu(cpu) {
+		q = per_cpu_ptr(ksyslog_queue, cpu);
+		seq_printf(seq, "  \"%u\": \"%lu\",\n", cpu, atomic64_read(&q->size));
+	}
+
+	seq_puts(seq, "}\n");
 	return 0;
 }
 
-static struct seq_operations ksyslog_queue_seq_ops = {
-	.start = ksyslog_rculist_seq_start,
-	.next  = ksyslog_rculist_seq_next,
-	.stop  = ksyslog_rculist_seq_stop,
-	.show  = ksyslog_queue_seq_show,
-};
-
-static int
-ksyslog_queue_seq_open(struct inode *inode, struct file *file)
-{
-	int err = seq_open(file, &ksyslog_queue_seq_ops);
-
-	if (!err)
-		((struct seq_file *)file->private_data)->private = PDE_DATA(inode);
-
-	return err;
-}
-
-static struct file_operations ksyslog_queue_fops = {
-	.owner   = THIS_MODULE,
-	.open    = ksyslog_queue_seq_open,
-	.read    = seq_read,
-	.llseek  = seq_lseek,
-	.release = seq_release,
-};
-
-static int
-ksyslog_size_seq_show(struct seq_file *seq, void *v)
-{
-	seq_printf(seq, "%lu\n", atomic64_read(&ksyslog_queue.size));
-	return 0;
-}
-
 static int
 ksyslog_size_seq_open(struct inode *inode, struct file *file)
@@ -487,44 +418,29 @@
 ksyslog_stats_seq_show(struct seq_file *seq, void *v)
 {
-	int i;
-	struct ksyslog_stats stats;
-
-	memset(&stats, 0, sizeof(stats));
-
-	for_each_possible_cpu(i) {
-		const struct ksyslog_stats *percpu_stats;
-		struct ksyslog_stats local_stats;
-		unsigned int start;
-
-		percpu_stats = per_cpu_ptr(ksyslog_queue.stats, i);
-
-		do {
-			start = u64_stats_fetch_begin_bh(&percpu_stats->sync);
-			local_stats = *percpu_stats;
-		} while (u64_stats_fetch_retry_bh(&percpu_stats->sync, start));
-
-		stats.write_bytes += local_stats.write_bytes;
-		stats.write_packets += local_stats.write_packets;
-		stats.drop_bytes += local_stats.drop_bytes;
-		stats.drop_packets += local_stats.drop_packets;
-		stats.discard_bytes += local_stats.discard_bytes;
-		stats.discard_packets += local_stats.discard_packets;
-	}
-
-	seq_puts(seq,   "{\n");
-	seq_puts(seq,   "  \"write\": {\n");
-	seq_printf(seq, "    \"bytes\":   \"%llu\",\n", stats.write_bytes);
-	seq_printf(seq, "    \"packets\": \"%llu\"\n", stats.write_packets);
-	seq_puts(seq,   "  },\n");
-	seq_puts(seq,   "  \"drop\": {\n");
-	seq_printf(seq, "    \"bytes\":   \"%llu\",\n", stats.drop_bytes);
-	seq_printf(seq, "    \"packets\": \"%llu\"\n", stats.drop_packets);
-	seq_puts(seq,   "  },\n");
-	seq_puts(seq,   "  \"discard\": {\n");
-	seq_printf(seq, "    \"bytes\":   \"%llu\",\n", stats.discard_bytes);
-	seq_printf(seq, "    \"packets\": \"%llu\"\n", stats.discard_packets);
-	seq_puts(seq,   "  }\n");
-	seq_puts(seq,   "}\n");
-
+	int cpu;
+	struct ksyslog_queue *q;
+
+	seq_puts(seq, "{\n");
+
+	for_each_possible_cpu(cpu) {
+		q = per_cpu_ptr(ksyslog_queue, cpu);
+
+		seq_printf(seq, "  \"%u\": {\n", cpu);
+		seq_puts(seq,   "    \"write\": {\n");
+		seq_printf(seq, "      \"bytes\":   \"%lu\",\n", atomic64_read(&q->write_stats.bytes));
+		seq_printf(seq, "      \"packets\": \"%lu\",\n", atomic64_read(&q->write_stats.packets));
+		seq_puts(seq,   "    },\n");
+		seq_puts(seq,   "    \"drop\": {\n");
+		seq_printf(seq, "      \"bytes\":   \"%lu\",\n", atomic64_read(&q->drop_stats.bytes));
+		seq_printf(seq, "      \"packets\": \"%lu\",\n", atomic64_read(&q->drop_stats.packets));
+		seq_puts(seq,   "    },\n");
+		seq_puts(seq,   "    \"discard\": {\n");
+		seq_printf(seq, "      \"bytes\":   \"%lu\",\n", atomic64_read(&q->discard_stats.bytes));
+		seq_printf(seq, "      \"packets\": \"%lu\",\n", atomic64_read(&q->discard_stats.packets));
+		seq_puts(seq,   "    },\n");
+		seq_puts(seq,   "  },\n");
+	}
+
+	seq_puts(seq, "}\n");
 	return 0;
 }
@@ -555,8 +471,4 @@
 ksyslog_proc_destroy(void)
 {
-	if (ksyslog_proc_queue)
-		remove_proc_entry("queue", ksyslog_procdir);
-	ksyslog_proc_queue = NULL;
-
 	if (ksyslog_proc_size)
 		remove_proc_entry("size", ksyslog_procdir);
@@ -576,13 +488,6 @@
 {
 	ksyslog_procdir = proc_mkdir("ksyslog", NULL);
-	if (ksyslog_procdir == NULL) {
+	if (!ksyslog_procdir) {
 		pr_err("ksyslog: proc_mkdir failed\n");
-		goto err;
-	}
-
-	ksyslog_proc_queue = proc_create_data("queue", S_IRUGO, ksyslog_procdir,
-					      &ksyslog_queue_fops, &ksyslog_queue.head);
-	if (ksyslog_proc_queue == NULL) {
-		pr_err("ksyslog: proc_create(queue) failed\n");
 		goto err;
 	}
@@ -590,5 +495,5 @@
 	ksyslog_proc_size = proc_create("size", S_IRUGO, ksyslog_procdir,
 					&ksyslog_size_fops);
-	if (ksyslog_proc_size == NULL) {
+	if (!ksyslog_proc_size) {
 		pr_err("ksyslog: proc_create(size) failed\n");
 		goto err;
@@ -597,5 +502,5 @@
 	ksyslog_proc_stats = proc_create("stats", S_IRUGO, ksyslog_procdir,
 					 &ksyslog_stats_fops);
-	if (ksyslog_proc_stats == NULL) {
+	if (!ksyslog_proc_stats) {
 		pr_err("ksyslog: proc_create(stats) failed\n");
 		goto err;
@@ -613,12 +518,12 @@
 ksyslog_finish(void)
 {
+	int cpu;
+
 	if (ksyslog_rcv_sk)
 		sock_release(ksyslog_rcv_sk);
 	ksyslog_rcv_sk = NULL;
 
-	if (ksyslog_wq) {
-		ksyslog_work_unregister();
+	if (ksyslog_wq)
 		destroy_workqueue(ksyslog_wq);
-	}
 	ksyslog_wq = NULL;
 
@@ -627,8 +532,9 @@
 #endif
 
-	ksyslog_entry_destroy(&ksyslog_queue);
+	for_each_possible_cpu(cpu)
+		ksyslog_entry_destroy(per_cpu_ptr(ksyslog_queue, cpu));
 	rcu_barrier();
 
-	ksyslog_queue_uninit(&ksyslog_queue);
+	ksyslog_queue_uninit();
 }
 
@@ -639,5 +545,5 @@
 	struct sockaddr_in sin;
 
-	err = ksyslog_queue_init(&ksyslog_queue);
+	err = ksyslog_queue_init(ksyslog_work_handler);
 	if (err)
 		goto err;
@@ -650,11 +556,9 @@
 
 	ksyslog_wq = create_workqueue("ksyslog");
-	if (ksyslog_wq == NULL) {
+	if (!ksyslog_wq) {
 		pr_err("ksyslog: create_workqueue failed\n");
 		err = -ENOMEM;
 		goto err;
 	}
-
-	INIT_DELAYED_WORK(&ksyslog_work, ksyslog_work_handler);
 
 	err = sock_create(AF_INET, SOCK_DGRAM, 0, &ksyslog_rcv_sk);
@@ -675,6 +579,4 @@
 	}
 
-	ksyslog_work_register(ksyslog_flush_interval);
-
 	udp_sk(ksyslog_rcv_sk->sk)->encap_type = UDP_ENCAP_KSYSLOG;
 	udp_sk(ksyslog_rcv_sk->sk)->encap_rcv = ksyslog_rcv;
