process: Introduce notify_job_request()

This function can be used to kick the main thread out of
poll_job_request() function in case someone wants to re-schedule rrd
processing.

As a starter, a notify call is added at the end of worker processing
when the last worker thread exits. The only purpose of this feature is
to make the main thread to print out the time of the next scheduled
job to be processed, making it just easier to follow the daemon as it
operates.

Signed-off-by: Timo Kokkonen <timo.t.kokkonen@iki.fi>

diff --git a/process.c b/process.c
index 8a4059bb34d44fc0f220902474d2ca92c77c616f..e2126a346d5bbde612400ac0351149cdde2e7d5d 100644 (file)
--- a/process.c
+++ b/process.c
@@ -53,6 +53,8 @@ struct mutex work_stats_mutex = {
 static int workers_active;
 static int worker_count;
 
 static int workers_active;
 static int worker_count;
 

+static int job_notify_fd[2];
+

 static int run_work_on_queue(struct work_queue *queue)
 {
        struct work_struct *work;
 static int run_work_on_queue(struct work_queue *queue)
 {
        struct work_struct *work;


@@ -132,9 +134,16 @@ out_unlock:
         * ensures next time we start spawning worker threads
         * the first thread will have number zero on its name.
         */
         * ensures next time we start spawning worker threads
         * the first thread will have number zero on its name.
         */

-       if (!workers_active)
+       if (!workers_active) {

                worker_count = 0;
 
                worker_count = 0;
 

+               /*
+                * Kick the job poller, just to print the time of next
+                * update on the logs
+                */
+               notify_job_request();
+       }
+

        mutex_unlock(&work_stats_mutex);
 
        return NULL;
        mutex_unlock(&work_stats_mutex);
 
        return NULL;


@@ -196,6 +205,18 @@ int queue_work(unsigned int priority, char *name,
        return 0;
 }
 
        return 0;
 }
 

+static int job_notify_handler(struct event_handler *h)
+{
+       int ret;
+       char buf[64];
+
+       ret = read(job_notify_fd[0], buf, sizeof(buf));
+       if (ret < 0)
+               pr_err("Failed to read: %m\n");
+
+       return 0;
+}
+

 /*
  * Initialize the jobcontrol.
  *
 /*
  * Initialize the jobcontrol.
  *


@@ -205,6 +226,7 @@ int queue_work(unsigned int priority, char *name,
  */
 int init_jobcontrol(int max_jobs_requested)
 {
  */
 int init_jobcontrol(int max_jobs_requested)
 {

+       static struct event_handler ev;

        FILE *file;
        int ret;
        char buf[256];
        FILE *file;
        int ret;
        char buf[256];


@@ -245,6 +267,18 @@ int init_jobcontrol(int max_jobs_requested)
                        max_jobs++;
        }
 
                        max_jobs++;
        }
 

+       ret = pipe(job_notify_fd);
+       if (ret) {
+               pr_err("pipe() failed: %m\n");
+               return ret;
+       }
+
+       ev.fd = job_notify_fd[0];
+       ev.events = EPOLLIN;
+       ev.handle_event = job_notify_handler;
+       ev.name = "job_notify";
+       register_event_handler(&ev, EPOLL_CTL_ADD);
+

 open_fail:
 read_fail:
        fclose(file);
 open_fail:
 read_fail:
        fclose(file);


@@ -304,6 +338,18 @@ out:
        return ret;
 }
 
        return ret;
 }
 

+int notify_job_request(void)
+{
+       int ret;
+       char byte = 0;
+
+       ret = write(job_notify_fd[1], &byte, sizeof(byte));
+       if (ret < 0)
+               pr_err("Failed to write: %m\n");
+
+       return 0;
+}
+

 int do_fork(void)
 {
        int child;
 int do_fork(void)
 {
        int child;

diff --git a/process.h b/process.h
index 1b92e5ba621076168b2be20cf413474ce0e40c25..705ef889341bb486f020341b79e366dc32ac69c5 100644 (file)
--- a/process.h
+++ b/process.h
@@ -27,6 +27,7 @@ int register_event_handler(struct event_handler *handler, int op);
 
 int init_jobcontrol(int max_jobs_requested);
 int poll_job_requests(int timeout);
 
 int init_jobcontrol(int max_jobs_requested);
 int poll_job_requests(int timeout);

+int notify_job_request(void);

 int do_fork(void);
 int run(const char *p, char *const argv[]);
 int clear_zombie(int pid);
 int do_fork(void);
 int run(const char *p, char *const argv[]);
 int clear_zombie(int pid);