Альтернативная функция сна для python.
Примечание. Не следует использовать для нескольких потоков из-за блокировки GIL, но для нескольких подпроцессов это нормально.То же самое с time.sleep ()
Я обертываю функцию C в Python.Я использую nanosleep () из библиотеки C, которая останавливает работу этого потока в течение такого большого количества времени.Это не задержка с занятым ожиданием, которая использует много ЦП для оценки математики.Коды следующие.Поместите все в папку, скажем, CWrapper.
C_functions.h
#include <time.h>
int c_sleep_msec(long milliseconds);
int c_sleep_nsec(long nanoseconds);
C_functions.c
#include "C_functions.h"
int c_sleep_msec(long milliseconds) {
struct timespec req;
//struct timespec rem;
if(milliseconds > 999) {
req.tv_sec = (int)(milliseconds / 1000); /* Must be Non-Negative */
req.tv_nsec = (milliseconds - ((long)req.tv_sec * 1000)) * 1000000; /* Must be in range of 0 to 999999999 */
}
else {
req.tv_sec = 0; /* Must be Non-Negative */
req.tv_nsec = milliseconds * 1000000; /* Must be in range of 0 to 999999999 */
}
//rem = NULL;
return nanosleep(&req , NULL);
}
//------------------------------------------------------
int c_sleep_nsec(long nanoseconds) {
struct timespec req;
//struct timespec rem;
if (nanoseconds > 999999999) {
req.tv_sec = (int)(nanoseconds / 1000000000);
req.tv_nsec = (nanoseconds - ((long)req.tv_sec * 1000000000));
}
else {
req.tv_sec = 0;
req.tv_nsec = nanoseconds;
}
//rem = NULL;
return nanosleep(&req , NULL);
}
Вы также можете создать функцию за микросекунды, используятот же nanosleep ()
CWrapper.pyx
cdef extern from "C_functions.h":
int c_sleep_msec(long milliseconds)
int c_sleep_nsec(long nanoseconds)
def sleep_msec(milliseconds):
return c_sleep_msec(milliseconds)
def sleep_nsec(nanoseconds):
return c_sleep_nsec(nanoseconds)
setup.py
from distutils.core import setup
from distutils.extension import Extension
from Pyrex.Distutils import build_ext
setup(
name = "CWrapper",
ext_modules=[ Extension("CWrapper", ["CWrapper.pyx", "C_functions.c"]) ],
cmdclass = {'build_ext': build_ext}
)
Установить Python-Pyrex.Затем запустите в linux-терминале
python setup.py build_ext -i
. Он создаст файлы CWrapper.c, build и CWrapper.so.Используйте CWrapper.so, где хотите, и просто импортируйте в python.
Примечание: скомпилируйте отдельно для Raspberry Pi.
Теперь протестируйте функцию
Test_sleep.py
import serial
from multiprocessing import Process
import time
import CWrapper
class TestSleep:
def __init__(self):
self.delay_sec = 0.00000100
self.delay_msec = 30
self.delay_nsec = 1000 #200000000
self.start_time = time.time()
self.process_1 = Process(name="process_1", target=self.process_1_task, args=("process_1",))
self.process_1.daemon = True
self.process_1.start()
self.process_2 = Process(name="process_2", target=self.process_1_task, args=("process_2",))
self.process_2.daemon = True
self.process_2.start()
self.process_3 = Process(name="process_3", target=self.process_1_task, args=("process_3",))
self.process_3.daemon = True
self.process_3.start()
def process_1_task(self, process_name):
start = self.start_time
delay_msec = self.delay_msec
delay_sec = self.delay_sec
delay_nsec = self.delay_nsec
t1 = start
for i in range(1, 81):
status = CWrapper.sleep_msec(delay_msec)
# status = CWrapper.sleep_nsec(delay_nsec)
#status = time.sleep(delay_sec)
t2 = time.time()
elapsed_time = t2 - t1
t1 = t2
print process_name, i, "status:", status, "Elapsed-time:", elapsed_time
if __name__ == '__main__':
test = TestSleep()
# for i in range(1,10000):
# print "main thread", i
# time.sleep(0.1)
while True: # Since daemon=True, main thread should check join() or stay in loop
pass
Изменение параметров delay_sec для time.sleep (), delay_msec для CWrapper.sleep_msec (), delay_nsec для CWrapper.sleep_nsec ().Раскомментируйте функцию, которую вы хотите протестировать в thread_1_task ().