U 7feU@s ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl Z ddl Z ddl ZGdddZddZddZd d ZGd d d e jZGd ddeZGdddeZGdddeZGdddeZGdddeZdddZddZdS)N)update_wrapperc@sReZdZiZeddZddZdddZd d Zdd d Z d dZ dddZ dS)CmfDeferredJobWrappercCs|jd|jS)N.) __module____name__funcr ./cmf/cmf_deferred_job.py gen_task_namesz#CmfDeferredJobWrapper.gen_task_namecCs|jjd|jdS)Nz(name=)) __class__rnameselfr r r __repr__szCmfDeferredJobWrapper.__repr__NFcCsJt||||_|||_||_||_||_||_||_||j |j<dSN) rrr rpriority system_job only_oncesoft_time_limit countdownregistry)rrrrrrrr r r __init__s  zCmfDeferredJobWrapper.__init__cOs |j||Srrrargskwargsr r r __call__)szCmfDeferredJobWrapper.__call__cCs$|dkr g}|dkri}|j||Srrrr r r apply,s zCmfDeferredJobWrapper.applycOs|j||dS)N)rr) apply_asyncrr r r delay3szCmfDeferredJobWrapper.delayc Cs2ddlm}m}ddl} |jrH|jj|jddrH|d|jddS|dkrTg}|dkr`i}|dkrn|j }|dkr||j }|dkr|j }|j|j|jd} |dk r|j | _ |s|j j | _|dkr|j}|rttj|d| _|||j| jj|j || j j|d d | _| jj| W5QRXt| j S) Nr)modelsgopen)rstatuszCmfDeferredJobWrapper(z): already queued)rr)Zseconds)r person_idrrrr)rroptions) cmf.includer!r" cmf.fieldsrCmfDeferredJobcountrdebugrrrZcurrent_personidr%rcmf_nowdatetime timedeltaplan_start_datetimevalue params_jsonutilcmfutilZ disable_aclsavestr) rrrrrrrr!r"cmfjobr r r r6sL  z!CmfDeferredJobWrapper.apply_async)NFNFN)NN)NNNNNN) rr __qualname__r staticmethodr rrrrr rr 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_rr r wrapperosz!cmf_deferred_job..wrapperNrz+cmf_deferred_job takes exactly 1 argument (z given)len TypeError)rrr<rr r;r cmf_deferred_jobjs   r@cCsddl}|jjjS)Nr)r(fieldsZ CmfDateTimenowr1)r7r r r r-sr-cCsZt}|d}t|d|d|d||}||dddkrV|dd}|S)zfrom logging Formatter classrr=N )ioStringIO tracebackprint_exceptiongetvalueclose)ZeiZsiotbsr r r format_exceptions rNcs2eZdZeddZdfdd ZddZZS) JobWorkerGreenletcCstjtjjSr)r.rBtimezoneutcr r r r rBszJobWorkerGreenlet.nowNcs||_tdSr) worker_idsuperr)rrRr r r rszJobWorkerGreenlet.__init__cCs|jjd|jdS)N(r )r rrRrr r r rszJobWorkerGreenlet.__repr__)N)rrr9r:rBrr __classcell__r r rTr rOs rOcsbeZdZdZdedfdd ZddZdd Zd d Zdd d Z ddZ ddZ ddZ Z S)QueueProcessorzJobQueueProcessor:get:lockN)rc s||_tjf|dSr)rrSr)rrrrTr r rszQueueProcessor.__init__c Csddlm}m}m}ddl}|jjj|jddd|j dddgdd ddgd d t ggg}|j dk r||d d|j gg}|j j|d gd |d}|rd|_|j|j|_|W5QRW5QRqW5QRXW5QRXtdq|d|j|S)Nr)r!socketio cmf_commit timeoutZblocking_timeoutr$=r#ORr0z<=rZcmf_created_atT)filterZorder_byZ for_updaterA in_progressrCzcmf-background-task-start-)r'r!rXrYcmf.appr3r4CmfLock _GET_LOCK_KEYapp cmf_contextr-rr)getr$start_datetimeset_nowrRr5timesleepemitr,)rrAr!rXrYr7Z job_filterZnext_jobr r r _job_get_nexts4$   * zQueueProcessor._job_get_nextc Cszddlm}ddl}|jRd|_|jt|j|j  |_ ||_ | |d|jt|W5QRXdS)NrrXsuccesszcmf-background-task-result-)r'rXrardrer$ end_datetimerhintrg total_secondsduration result_jsonr5rkr,r6)rr8resultrXr7r r r _job_set_successs   zQueueProcessor._job_set_successc Csddlm}ddl}|jhd|_|jt|j|j  |_ d|_ t }|drft||_||d|j|W5QRXdS)Nrrmfailzcmf-background-task-error-)r'rXrardrer$rorhrprgrqrrrssysexc_inforNZ error_textr5rkr,)rr8errorrXr7rxr r r _job_set_fails    zQueueProcessor._job_set_failcs0ddlmddlfdd}t||S)Nr)r!c sttjj}jTjrHjjjjjj j dj |j dj dW5QRSQRXdS)NrArr)rrrrdrer%Zset_current_personZ CmfPersonrfAPPZcurrent_person_fieldsZ CmfAccessListZ setup_contextr2)job_funcr7r8r!r r runners   z'QueueProcessor._job_run..runner)r'r!rageventZ with_timeout)rr8rrr r~r _job_runs  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$rRr2Z param_pickler%r{r&r)rz job_runner(z, ) error)rlr2rfrrur GreenletExit SystemExitKeyboardInterrupt Exceptionlogging exceptionr,rrzr6)rr8rrter r r _run_cycleszQueueProcessor._run_cyclec Cshz|tdWqtttjfk r4Yqtk r`t |dtdYqXqdS)zmain() for Greenletr=z cycle error.N) rrirjrrrrrrrrr r r _runszQueueProcessor._runcCs|jjd|jd|jdS)NrUz , priority=r )r rrRrrr r r rszQueueProcessor.__repr__)N)N)rrr9rcrprrlrurzrrrrrVr r rTr rWs   rWc@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!,r[)ZdaysZhoursr^r$INrvcanceldeadroOsz%QueueWatcher._run..r$r]r`rgrrr<) rar'rr!rdr|REDIS_DBredisr3r4rbrreZsmemberslistrfrZsremremoveprintrwstderrr-r.r/r)Z bulk_updaterrrrirj) rr7rr!redis_dbZ db_workersrRrBZmax_start_dater r r r6s6 $     zQueueWatcher._runNrr r r r r3src@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) rardrecallable isinstancerrryrrrrr)rZhooksr7hookr}paramsr r r _schedule_hooksgs(   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)rardr|rrrBhourrandomZrandintrirjr.rPrQsetrRrHOOK_CRON_MINUTELYminuteHOOK_CRON_HOURLYrrr)rr7rZ hourly_lastZ hourly_run_atrBr r r rs    zJobScheduler._runN)rrr9__doc__rrr r r r rZs &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_CACHErrC2r[redis_idi`Tz%RedisCleaner: run CMF_CACHE.flushdb()rr)rar'rrrdr|rrrirjr.rBrrr3r4rbrrfrrwrZflushdbrrr)rr7rrrrBrr r r rs$    zRedisCleaner._runN)rrr9rrrr 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:)rrR )rRr-)rrz$deferred_job_worker: redis error!!! rZ)r\z%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.)"raplatformZnodeuuidZuuid1rdr|rrappendrWrangerrrrstartrZsaddrrrrZiwaitrrwrrrrZkillallZTimeoutZ successfulrf) Z single_queuer7rRrZ greenletsrglrZgl_iterr r r deferred_job_workersR   "(  rcCs0ddl}dd}||jjj||jjjdS)NrcSsrt||D]`}t|r|}n|\}}t|ts@td|d|jsVtd|d|js td|dq dS)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.)rrrr ValueErrorrr)Z hook_listrr}Z_paramsr r r check_hook_list s z0check_job_configuration..check_hook_list)rardr|rr)r7rr r r check_job_configurationsr)F)r.rFrrirHrwrrr functoolsrrrZcmf.util.cmfutilr7rr@r-rNZGreenletrOrWrrrrrrr r r r s0 X u'N C