U [SeT@s ddlZddlZddlZddlZddlZddlZddlZddlZddlZddl m Z ddl Z ddl Z ddl ZGdddZddZddZd d ZGd d d e jZGd ddeZGdddeZGdddeZGdddeZGdddeZdddZddZdS)N)update_wrapperc@sHeZdZiZeddZddZdddZd d Zd d Z dd dZ 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__cOs|j||dS)N)rr) apply_asyncrr r r delay,szCmfDeferredJobWrapper.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 r/sL  z!CmfDeferredJobWrapper.apply_async)NFNFN)NNNNNN) rr __qualname__r staticmethodr rrrrrr 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 wrapperhsz!cmf_deferred_job..wrapperNrz+cmf_deferred_job takes exactly 1 argument (z given)len TypeError)rrr;rr r:r cmf_deferred_jobcs   r?cCsddl}|jjjS)Nr)r'fieldsZ CmfDateTimenowr0)r6r r r r,xsr,cCsZt}|d}t|d|d|d||}||dddkrV|dd}|S)zfrom logging Formatter classrr<N )ioStringIO tracebackprint_exceptiongetvalueclose)ZeiZsiotbsr r r format_exception~s rMcs2eZdZeddZdfdd ZddZZS) JobWorkerGreenletcCstjtjjSr)r-rAtimezoneutcr r r r rAszJobWorkerGreenlet.nowNcs||_tdSr) worker_idsuperr)rrQr r r rszJobWorkerGreenlet.__init__cCs|jjd|jdS)N(r )r rrQrr r r rszJobWorkerGreenlet.__repr__)N)rrr8r9rArr __classcell__r r rSr rNs rNcsbeZdZdZdedfdd ZddZdd Zd d Zdd d Z ddZ ddZ ddZ Z S)QueueProcessorzJobQueueProcessor:get:lockN)rc s||_tjf|dSr)rrRr)rrrrSr 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"ORr/z<=rZcmf_created_atT)filterZorder_byZ for_updater@ in_progressrBzcmf-background-task-start-)r&r rWrXcmf.appr2r3CmfLock _GET_LOCK_KEYapp cmf_contextr,rr(getr#start_datetimeset_nowrQr4timesleepemitr+)rr@r rWrXr6Z job_filterZnext_jobr r r _job_get_nexts4$   * zQueueProcessor._job_get_nextc Cszddlm}ddl}|jRd|_|jt|j|j  |_ ||_ | |d|jt|W5QRXdS)NrrWsuccesszcmf-background-task-result-)r&rWr`rcrdr# end_datetimergintrf total_secondsduration result_jsonr4rjr+r5)rr7resultrWr6r r r _job_set_successs   zQueueProcessor._job_set_successc Csddlm}ddl}|jhd|_|jt|j|j  |_ d|_ t }|drft||_||d|j|W5QRXdS)Nrrlfailzcmf-background-task-error-)r&rWr`rcrdr#rnrgrorfrprqrrsysexc_inforMZ error_textr4rjr+)rr7errorrWr6rwr r r _job_set_fails    zQueueProcessor._job_set_failc Csddlm}ddl}tj|j}|j^|jr\|j |j j |j|jj j d|jtj||f|jd|jdW5QRSQRXdS)Nr)r r@rr)r&r r`rrrrcrdr$Zset_current_personZ CmfPersonreAPPZcurrent_person_fieldsZ CmfAccessListZ setup_contextgeventZ with_timeoutr1)rr7rr r6job_funcr 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#rQr1Z param_pickler$rzr%r)rz job_runner(z, ) error)rkr1rer~rtr| GreenletExit SystemExitKeyboardInterrupt Exceptionlogging exceptionr+rryr5)rr7rrser r r _run_cycleszQueueProcessor._run_cyclec Cshz|tdWqtttjfk r4Yqtk r`t |dtdYqXqdS)zmain() for Greenletr<z cycle error.N) rrhrirrr|rrrrrr r r _runszQueueProcessor._runcCs|jjd|jd|jdS)NrTz , priority=r )r rrQrrr r r r szQueueProcessor.__repr__)N)N)rrr8rbrorrkrtryr~rrrrUr r rSr rVs   rVc@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 ,rZ)ZdaysZhoursr]r#INrucanceldeadrnHsz%QueueWatcher._run..r#r\r_rfrrr<) r`r&rr rcr{REDIS_DBredisr2r3rarrdZsmemberslistrerZsremremoveprintrvstderrr,r-r.r(Z bulk_updaterrrrhri) rr6rr redis_dbZ db_workersrQrAZmax_start_dater r r r/s6 $     zQueueWatcher._runNrr r r r r,src@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) r`rcrdcallable isinstancerrrxrrrrr)rZhooksr6hookr}paramsr r r _schedule_hooks`s(   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)r`rcr{rrrAhourrandomZrandintrhrir-rOrPsetrQrHOOK_CRON_MINUTELYminuteHOOK_CRON_HOURLYrrr)rr6rZ hourly_lastZ hourly_run_atrAr r r rs    zJobScheduler._runN)rrr8__doc__rrr r r r rSs &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_CACHErrB2rZredis_idi`Tz%RedisCleaner: run CMF_CACHE.flushdb()rr)r`r&rrrcr{rrrhrir-rArrr2r3rarrerrvrZflushdbrrr)rr6rrrrArr r r rs$    zRedisCleaner._runN)rrr8rrrr 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:)rrQ )rQr-)rrz$deferred_job_worker: redis error!!! rY)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.)"r`platformZnodeuuidZuuid1rcr{rrappendrVrangerrrrstartrZsaddrrrr|ZiwaitrrvrrrrZkillallZTimeoutZ successfulre) Z single_queuer6rQrZ 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_lists z0check_job_configuration..check_hook_list)r`rcr{rr)r6rr r r check_job_configurationsr)F)r-rErrhrGrvrrr functoolsrr|rZcmf.util.cmfutilr6rr?r,rMZGreenletrNrVrrrrrrr r r r s0 Q u'N C