U |j}|dkrL|j}|dkrZ|j}|j|j|j d} |dk r||j| _|s|j j | _ |j rbt |j} d} |jdkr|rtd|jd|f| t |7} r| t fddtD7} n|jD]} | t | 7} q| r2| d t| 7} | | _|jj| d d rb|d |jd dS|dkrr|j}|rttj|d| _||j |j| j j|j|| jj||jdd| _| j j!"| #W5QRXt | j S)Nr)modelsg)rrzJCmfDeferredJobWrapper.apply_async: !!! Warning! Use args + only_once! job=z, args=csi|]}||qSr r ).0krr r ]sz5CmfDeferredJobWrapper.apply_async..__open) only_once_keystatuszCmfDeferredJobWrapper(z): already queued)Zseconds)rr person_idrrrr admin_mode)rroptions)$ cmf.includer"r# cmf.fieldsrrrCmfDeferredJobrrZcurrent_personidr-strrprintsortedhashlibZmd5encodeZ hexdigestr+countdebugrcmf_nowdatetime timedeltaplan_start_datetimevalueZacl_admin_mode params_jsonutilcmfutilZ disable_aclsave)rrrrrrrr"r#cmfjobr+args_strZarg_keyr r'r r :sn       z!CmfDeferredJobWrapper.apply_async)NFNNFN)NN)NNNNNN) rr __qualname__r staticmethodr rrrrr!r r r r r rs   rcsNfdd}d}|r:t|dkr*|d}ntd|d|rF||S|SdS)za job decorator: return object like celery task with __call__, delay, apply_async cs t|fSr)r)Zfunc_r'r r wrappersz!cmf_deferred_job..wrapperNrz+cmf_deferred_job takes exactly 1 argument (z given)len TypeError)rrrIrr r'r cmf_deferred_jobs   rMcCsddl}|jjjS)Nr)r1fieldsZ CmfDateTimenowr?)rDr r r r;sr;cCsZt}|d}t|d|d|d||}||dddkrV|dd}|S)zfrom logging Formatter classrrJN )ioStringIO tracebackprint_exceptiongetvalueclose)ZeiZsiotbsr r r format_exceptions r[cs2eZdZeddZdfdd ZddZZS) JobWorkerGreenletcCstjtjjSr)r<rOtimezoneutcr r r r rOszJobWorkerGreenlet.nowNcs||_tdSr) worker_idsuperr)rr_r r r rszJobWorkerGreenlet.__init__cCs|jjd|jdS)N(r )r rr_rr r r rszJobWorkerGreenlet.__repr__)N)rrrGrHrOrr __classcell__r r rar r\s r\csbeZdZdZdedfdd ZddZdd Zd d Zdd d Z ddZ ddZ ddZ Z S)QueueProcessorzJobQueueProcessor:get:lockN)rc s||_tjf|dSr)rr`r)rrrrar r rszQueueProcessor.__init__c Cs^ddlm}m}m}ddl}|jjj|jddd|j dddgdd ddgd d t ggg}|j dk r~|d d|j gg}|j jd dd gdddgdddggddgd}|r|ddddgdddd|Dgg|j j|dgd |d}|r(d|_|j|j|_|W5QRW5QRqHW5QRXW5QRXtdq|d|j|S)Nr)r"socketio cmf_commit timeoutZblocking_timeoutr,=r*ORr>z<=rrTr+z!= in_progressz--)filterrNNOT INcSsg|] }|jqSr )r+)r%Zjob_r r r sz0QueueProcessor._job_get_next..Zcmf_created_at)rmZorder_byZ for_updaterNrPzcmf-background-task-start-)r0r"rerfcmf.apprArBCmfLock _GET_LOCK_KEYapp cmf_contextr;rr2Zslistappendgetr,start_datetimeset_nowr_rCtimesleepemitr3) rrNr"rerfrDZ job_filterZonly_once_in_progressZnext_jobr r r _job_get_nextsJ&   , zQueueProcessor._job_get_nextc Cszddlm}ddl}|jRd|_|jt|j|j  |_ ||_ | |d|jt|W5QRXdS)Nrresuccesszcmf-background-task-result-)r0rerprsrtr, end_datetimerxintrw total_secondsduration result_jsonrCr{r3r4)rrEresultrerDr r r _job_set_successs   zQueueProcessor._job_set_successc Csddlm}ddl}|jhd|_|jt|j|j  |_ d|_ t }|drft||_||d|j|W5QRXdS)Nrr}failzcmf-background-task-error-)r0rerprsrtr,rrxrrwrrrsysexc_infor[Z error_textrCr{r3)rrEerrorrerDrr r r _job_set_fails    zQueueProcessor._job_set_failcs0ddlmddlfdd}t||S)Nr)r"c stjj}jpjrdjjjjjj j dj j dddrdj |j dj dW5QRSQRXdS)NrNr/r.Frr)rrrrsrtr-Zset_current_personZ CmfPersonrvAPPZcurrent_person_fieldsZ CmfAccessListZ setup_contextr@Zactivate_admin_mode)job_funcrDrEr"r r runners    z'QueueProcessor._job_run..runner)r0r"rpgeventZ with_timeout)rrErrr rr _job_run s  zQueueProcessor._job_runc Cs|jdddddgd}|sdS|jdd}z|j||d }|||Wn|tjk rdYnfttfk r|YnNt k r}z0t d |j d |j d ||t|W5d}~XYnXdS) Nr,r_r@Z param_pickler-rr/r)rz job_runner(z, ) error)r|r@rvrrr GreenletExit SystemExitKeyboardInterrupt Exceptionlogging exceptionr3rrr4)rrErrer r r _run_cycleszQueueProcessor._run_cyclec Cshz|tdWqtttjfk r4Yqtk r`t |dtdYqXqdS)zmain() for GreenletrJz cycle error.N) rryrzrrrrrrrrr r r _run6szQueueProcessor._runcCs|jjd|jd|jdS)Nrbz , priority=r )r rr_rrr r r rCszQueueProcessor.__repr__)N)N)rrrGrrrrr|rrrrrrrcr r rar rds(   rdc@seZdZdZddZdS) QueueCleanerzJobQueueCleaner:lockc Csddl}ddlm}m}z|jjj|jddd||j ft }|t j dd}|t j dd}|j jd d d d d dggdd|ggd ddgdd|gggdW5QRXW5QRXWn6|k rYn$tk rt|dYnXtdqdS)NrCmfGetLockErrorr",rh)ZdaysZhoursrkr,INrcanceldeadr.r,rjrlrwrrr<) rpr0rr"rsrREDIS_DBredisrArBrqrrtZsmemberslistrvrZsremremover5rstderrr;r<r=r2Z bulk_updaterrrryrz) rrDrr"redis_dbZ db_workersr_rOZmax_start_dater r r rgs6 $     zQueueWatcher._runNrr r r r rdsrc@s eZdZdZddZddZdS) JobScheduleru - защита от параллельного запуска - "размазывание нагрузки" во времени. - HOOK_CRON_MINUTELY и HOOK_CRON_HOURLY в первый цикл не запускаем, т.е. HOOK_CRON_MINUTELY запускаем на второй минуте работы APP.HOOK_CRON_MINUTELY = [] APP.HOOK_CRON_HOURLY = [] TODO: APP.HOOK_CRON_DAILY = [] APP.HOOK_CRON_WEEKLY = [] APP.HOOK_CRON_MONTHLY = [] c Csddl}|j|D]}zt|r0|}i}n|\}}t|ts\t|d|dWq|js|t|d|dWq|j st|d|dWq|j f|Wqt k rt |d|dYqXqW5QRXdS) u hook - вероятно должен быть только CmfDeferredJob. Произвольные лёгкие функции можно выполнять без джобов в отдельном контексте и с лимитом времени, но вероятно такое не нужно, и будут злоупотреблять. Вторая форма, если нужно передать параметры: hook = (hook, {**apply_async_params(args=, kwargs=, priority=, ...)}) Есть мнение, что hook.only_once должен быть всегда True у периодических задач. Можно его выставлять принудительно, но наверное пока лучше падать, чтобы не забывали явно указывать в объявлении джоба. Так же пока будем требовать system_job=True rNz: periodic hook(z+) must be Job(@cmf_deferred_job). Skip run.z*) must be with option only_once. Skip run.z+) must be with option system_job. Skip run.z schedule hook(r) rprsrtcallable isinstancerrrrrr rr)rZhooksrDhookrparamsr r r _schedule_hookss(   zJobScheduler._schedule_hookscCsddl}|jjjj}|j}tdd}t dzt j t j j }|jd|d|jdddrr||jjj|j|kr|j|kr|j}|jd |d |jd ddr||jjjWq*tk rt|d Yq*Xq*dS) NrrzJobScheduler:minutely:z %Y%m%d_%H%MxT)exZnxzJobScheduler:hourly:z %Y%m%d_%Hi r)rprsrrrrOhourrandomZrandintryrzr<r]r^setr_rHOOK_CRON_MINUTELYminuteHOOK_CRON_HOURLYrrr)rrDrZ hourly_lastZ hourly_run_atrOr r r rs    zJobScheduler._runN)rrrG__doc__rrr r r r rs &rc@seZdZdZdZddZdS) RedisCleaneruDРаз в сутки сбрасываем редис БД. В 02-50.zRedisCleaner:lockc Csddl}ddlm}m}|jjjj}t dzt j }|j dksN|j dkrRWq$|jjj|jdddH|d}|rt|dkrW5QRWq$td tjd |W5QRXWq$|k rYq$tk rt|d Yq$Xq$dS) Nr)r CMF_CACHErrP2rhredis_idi`Tz%RedisCleaner: run CMF_CACHE.flushdb()rr)rpr0rrrsrrrryrzr<rOrrrArBrqrrvr5rrZflushdbrrr)rrDrrrrOrr r r rs$    zRedisCleaner._runN)rrrGrrrr r r r rsrFc Cslddl}tdt}|jjjj}g}|rF| t d|dn t dD]}| t ||dqN| t |d| t |d| t|d| t|d|D] }|qzz&|jd|dd d |d |Wn4tk r}ztd |W5d}~XYnXtj|d d,}|D] }td|dtjdtq*W5QRXqWnBtttjfk r}ztd|dtjdW5d}~XYnXztj|d dWn&tjk rtdtjdYnX|D]}|rtd|dtjdq| r$td|dtjdqz |!Wn2tttjtfk rbtd|dYnXqdS)u! Функция - обработчик очередей задач. На каждый приоритет(очередь) запускаем отдельный поток - обработчик. Сервисный поток для очистки очереди от устаревших задач. Сервисный поток для контроля погибших задач(чей воркер погиб). Сервисный поток Beat, для планирования периодических задач. rN:)rr_ )r_r-)rrz$deferred_job_worker: redis error!!! rg)riz%deferred_job_worker(): system thread(z) exited!!! Stop work.rzdeferred_job_worker(): stop by z. Try kill all threads...z(deferred_job_worker(): killall timeout!.zdeferred_job_worker(): Thread z is not stopped!!!z stops normally!z error.)"rpplatformZnodeuuidZuuid1rsrrrrurdrangerrrrstartrZsaddrrrrZiwaitr5rrrrrZkillallZTimeoutZ successfulrv) Z single_queuerDr_rZ greenletsrglrZgl_iterr r r deferred_job_workersR   "(  rcCs0ddl}dd}||jjj||jjjdS)NrcSsj|D]`}t|r|}n|\}}t|ts8td|d|jsNtd|d|jstd|dqdS)Nz'check_job_configuration: periodic hook(z%) must be Job(use @cmf_deferred_job).z ) must be with option only_once.z!) must be with option system_job.)rrr ValueErrorrr)Z hook_listrrZ_paramsr r r check_hook_list<s z0check_job_configuration..check_hook_list)rprsrrr)rDrr r r check_job_configuration9sr)F)r<r7rSrryrUrrrr functoolsrrrZcmf.util.cmfutilrDrrMr;r[ZGreenletr\rdrrrrrrr r r r s4 x 'N C