C based timer

From SoftwareGuy
Jump to navigation Jump to search

C Timer[edit | edit source]

#include <iostream>
#include <unistd.h>
using namespace std;

// Define the callback function

void callback() {
  cout << "Timer callback called!" << endl;
}

// Create the timer
timer_t timer;
struct sigevent sigev;
// Set up the timer

sigev.sigev_notify = SIGEV_THREAD;
sigev.sigev_notify_function = callback;
sigev.sigev_value.sival_ptr = &timer;

timer_create(CLOCK_REALTIME, &sigev, &timer);
// Start the timer
timer_settime(timer, 0, new itimerspec{1, 0}, NULL);
// Wait for the timer to expire
pause();
// Delete the timer
timer_delete(timer);

Python timer[edit | edit source]

import threading, time, signal

from datetime import timedelta

WAIT_TIME_SECONDS = 1

class ProgramKilled(Exception):
    pass
def foo():
    print time.ctime()
def signal_handler(signum, frame):
    raise ProgramKilled
class Job(threading.Thread):
    def __init__(self, interval, execute, *args, **kwargs):
        threading.Thread.__init__(self)
        self.daemon = False
        self.stopped = threading.Event()
        self.interval = interval
        self.execute = execute
        self.args = args
        self.kwargs = kwargs
        
    def stop(self):
        self.stopped.set()
        self.join()
   def run(self):
        while not self.stopped.wait(self.interval.total_seconds()):
            self.execute(*self.args, **self.kwargs)
if __name__ == "__main__":
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)
    job = Job(interval=timedelta(seconds=WAIT_TIME_SECONDS), execute=foo)
    job.start()
   
    while True:
          try:
              time.sleep(1)
          except ProgramKilled:
              print "Program killed: running cleanup code"
              job.stop()
              break

  1. output
  2. Tue Oct 16 17:47:51 2018
  3. Tue Oct 16 17:47:52 2018
  4. Tue Oct 16 17:47:53 2018

More examples[edit | edit source]

import datetime, threading, time

next_call = time.time()

def foo():
    global next_call
    print datetime.datetime.now()
    next_call = next_call+1
    threading.Timer( next_call - time.time(), foo ).start()

foo()


import datetime, threading, time

def foo():
    next_call = time.time()
    while True:
        print datetime.datetime.now()        
        next_call = next_call+1;
time.sleep(next_call - time.time())

timerThread = threading.Thread(target=foo)
timerThread.start()

Followed by:

timerThread = threading.Thread(target=foo)
timerThread.daemon = True
timerThread.start()

import threading
def work ():
    threading.Timer(0.25, work).start ()
    print "stackoverflow"
work ()


import threading
def do_every (interval, worker_func, iterations = 0):
 if iterations != 1:
   threading.Timer (
     interval,
     do_every, [interval, worker_func, 0 if iterations == 0 else iterations-1]
   ).start ()
 worker_func ()
def print_hw ():
 print "hello world"
def print_so ():
 print "stackoverflow"
# call print_so every second, 5 times total
do_every (1, print_so, 5)
# call print_hw two times per second, forever
do_every (0.5, print_hw)