Changeset - fb0c3af6031b
[Not reviewed]
celery
1 6 1
Marcin Kuzminski - 15 years ago 2010-09-23 01:08:33
marcin@python-works.com
Implemented locking for task, to prevent for running the same tasks,
moved out pidlock library. Added dirsize display
7 files changed with 60 insertions and 20 deletions:
0 comments (0 inline, 0 general)
pylons_app/lib/celerylib/__init__.py
Show inline comments
 
from pylons_app.lib.pidlock import DaemonLock, LockHeld
 
from vcs.utils.lazy import LazyProperty
 
from decorator import decorator
 
import logging
 
import os
 
import sys
 
import traceback
 

	
 
from hashlib import md5
 
log = logging.getLogger(__name__)
 

	
 
class ResultWrapper(object):
 
    def __init__(self, task):
 
        self.task = task
 
        
 
    @LazyProperty
 
    def result(self):
 
        return self.task
 

	
 
def run_task(task, *args, **kwargs):
 
    try:
 
        t = task.delay(*args, **kwargs)
 
        log.info('running task %s', t.task_id)
 
        return t
 
    except Exception, e:
 
        print e
 
        if e.errno == 111:
 
            log.debug('Unnable to connect. Sync execution')
 
        else:
 
            log.error(traceback.format_exc())
 
        #pure sync version
 
        return ResultWrapper(task(*args, **kwargs))
 
    
 

	
 
class LockTask(object):
 
    """LockTask decorator"""
 
    
 
    def __init__(self, func):
 
        self.func = func
 
        
 
    def __call__(self, func):
 
        return decorator(self.__wrapper, func)
 
    
 
    def __wrapper(self, func, *fargs, **fkwargs):
 
        params = []
 
        params.extend(fargs)
 
        params.extend(fkwargs.values())
 
        lockkey = 'task_%s' % \
 
           md5(str(self.func) + '-' + '-'.join(map(str, params))).hexdigest()
 
        log.info('running task with lockkey %s', lockkey)
 
        try:
 
            l = DaemonLock(lockkey)
 
            return func(*fargs, **fkwargs)
 
            l.release()
 
        except LockHeld:
 
            log.info('LockHeld')
 
            return 'Task with key %s already running' % lockkey   
 

	
 
            
 
            
 

	
 
        
 
        
 
    
 
    
 
    
 
  
pylons_app/lib/celerylib/tasks.py
Show inline comments
 
from celery.decorators import task
 
from celery.task.sets import subtask
 
from celeryconfig import PYLONS_CONFIG as config
 
from pylons.i18n.translation import _
 
from pylons_app.lib.celerylib import run_task
 
from pylons_app.lib.celerylib import run_task, LockTask
 
from pylons_app.lib.helpers import person
 
from pylons_app.lib.smtp_mailer import SmtpMailer
 
from pylons_app.lib.utils import OrderedDict
 
from operator import itemgetter
 
from vcs.backends.hg import MercurialRepository
 
from time import mktime
 
import traceback
 
import json
 

	
 
__all__ = ['whoosh_index', 'get_commits_stats',
 
           'reset_user_password', 'send_email']
 

	
 
@@ -59,49 +59,51 @@ def get_hg_ui_settings():
 
            k = k.replace('.', '_')
 
        
 
        if each.ui_section == 'hooks':
 
            v = each.ui_active
 
        
 
        settings[each.ui_section + '_' + k] = v  
 
    
 
    return settings   
 

	
 
@task
 
def whoosh_index(repo_location, full_index):
 
    log = whoosh_index.get_logger()
 
    from pylons_app.lib.indexers import DaemonLock
 
    from pylons_app.lib.pidlock import DaemonLock
 
    from pylons_app.lib.indexers.daemon import WhooshIndexingDaemon, LockHeld
 
    try:
 
        l = DaemonLock()
 
        WhooshIndexingDaemon(repo_location=repo_location)\
 
            .run(full_index=full_index)
 
        l.release()
 
        return 'Done'
 
    except LockHeld:
 
        log.info('LockHeld')
 
        return 'LockHeld'    
 

	
 

	
 
@task
 
@LockTask('get_commits_stats')
 
def get_commits_stats(repo_name, ts_min_y, ts_max_y):
 
    author_key_cleaner = lambda k: person(k).replace('"', "") #for js data compatibilty
 
        
 
    from pylons_app.model.db import Statistics, Repository
 
    log = get_commits_stats.get_logger()
 
    commits_by_day_author_aggregate = {}
 
    commits_by_day_aggregate = {}
 
    repos_path = get_hg_ui_settings()['paths_root_path'].replace('*', '')
 
    repo = MercurialRepository(repos_path + repo_name)
 

	
 
    skip_date_limit = True
 
    parse_limit = 500 #limit for single task changeset parsing
 
    parse_limit = 350 #limit for single task changeset parsing
 
    last_rev = 0
 
    last_cs = None
 
    timegetter = itemgetter('time')
 
    
 
    sa = get_session()
 
    
 
    dbrepo = sa.query(Repository)\
 
        .filter(Repository.repo_name == repo_name).scalar()
 
    cur_stats = sa.query(Statistics)\
 
        .filter(Statistics.repository == dbrepo).scalar()
 
    if cur_stats:
 
        last_rev = cur_stats.stat_on_revision
 
@@ -197,24 +199,26 @@ def get_commits_stats(repo_name, ts_min_
 
    stats.repository = dbrepo
 
    stats.stat_on_revision = last_cs.revision
 
    stats.languages = json.dumps({'_TOTAL_':0, '':0})
 
    
 
    try:
 
        sa.add(stats)
 
        sa.commit()    
 
    except:
 
        log.error(traceback.format_exc())
 
        sa.rollback()
 
        return False
 
                        
 
    run_task(get_commits_stats, repo_name, ts_min_y, ts_max_y)
 
                            
 
    return True
 

	
 
@task
 
def reset_user_password(user_email):
 
    log = reset_user_password.get_logger()
 
    from pylons_app.lib import auth
 
    from pylons_app.model.db import User
 
    
 
    try:
 
        try:
 
            sa = get_session()
 
            user = sa.query(User).filter(User.email == user_email).scalar()
pylons_app/lib/indexers/__init__.py
Show inline comments
 
from os.path import dirname as dn, join as jn
 
from pidlock import LockHeld, DaemonLock
 
from pylons_app.config.environment import load_environment
 
from pylons_app.model.hg_model import HgModel
 
from shutil import rmtree
 
from webhelpers.html.builder import escape
 
from vcs.utils.lazy import LazyProperty
 

	
 
from whoosh.analysis import RegexTokenizer, LowercaseFilter, StopFilter
 
from whoosh.fields import TEXT, ID, STORED, Schema, FieldType
 
from whoosh.index import create_in, open_dir
 
from whoosh.formats import Characters
 
from whoosh.highlight import highlight, SimpleFragmenter, HtmlFormatter   
 

	
pylons_app/lib/indexers/daemon.py
Show inline comments
 
@@ -23,25 +23,25 @@ Created on Jan 26, 2010
 
@author: marcink
 
A deamon will read from task table and run tasks
 
"""
 
import sys
 
import os
 
from os.path import dirname as dn
 
from os.path import join as jn
 

	
 
#to get the pylons_app import
 
project_path = dn(dn(dn(dn(os.path.realpath(__file__)))))
 
sys.path.append(project_path)
 

	
 
from pidlock import LockHeld, DaemonLock
 
from pylons_app.lib.pidlock import LockHeld, DaemonLock
 
from pylons_app.model.hg_model import HgModel
 
from pylons_app.lib.helpers import safe_unicode
 
from whoosh.index import create_in, open_dir
 
from shutil import rmtree
 
from pylons_app.lib.indexers import INDEX_EXTENSIONS, IDX_LOCATION, SCHEMA, IDX_NAME
 

	
 
import logging
 

	
 
log = logging.getLogger('whooshIndexer')
 
# create logger
 
log.setLevel(logging.DEBUG)
 
log.propagate = False
pylons_app/lib/pidlock.py
Show inline comments
 
file renamed from pylons_app/lib/indexers/pidlock.py to pylons_app/lib/pidlock.py
 
import os, time
 
import sys
 
from warnings import warn
 

	
 
class LockHeld(Exception):pass
 

	
 

	
 
class DaemonLock(object):
 
    '''daemon locking
 
    """daemon locking
 
    USAGE:
 
    try:
 
        l = lock()
 
        main()
 
        l.release()
 
    except LockHeld:
 
        sys.exit(1)
 
    '''
 
    """
 

	
 
    def __init__(self, file=None, callbackfn=None,
 
                 desc='daemon lock', debug=False):
 

	
 
        self.pidfile = file if file else os.path.join(os.path.dirname(__file__),
 
                                                      'running.lock')
 
        self.callbackfn = callbackfn
 
        self.desc = desc
 
        self.debug = debug
 
        self.held = False
 
        #run the lock automatically !
 
        self.lock()
 
@@ -31,27 +31,27 @@ class DaemonLock(object):
 
    def __del__(self):
 
        if self.held:
 

	
 
#            warn("use lock.release instead of del lock",
 
#                    category = DeprecationWarning,
 
#                    stacklevel = 2)
 

	
 
            # ensure the lock will be removed
 
            self.release()
 

	
 

	
 
    def lock(self):
 
        '''
 
        """
 
        locking function, if lock is present it will raise LockHeld exception
 
        '''
 
        """
 
        lockname = '%s' % (os.getpid())
 

	
 
        self.trylock()
 
        self.makelock(lockname, self.pidfile)
 
        return True
 

	
 
    def trylock(self):
 
        running_pid = False
 
        try:
 
            pidfile = open(self.pidfile, "r")
 
            pidfile.seek(0)
 
            running_pid = pidfile.readline()
 
@@ -66,48 +66,48 @@ class DaemonLock(object):
 
                        print "It is running as process %s" % running_pid
 
                        raise LockHeld
 
                else:
 
                        print "Lock File is there but the program is not running"
 
                        print "Removing lock file for the: %s" % running_pid
 
                        self.release()
 
        except IOError, e:
 
            if e.errno != 2:
 
                raise
 

	
 

	
 
    def release(self):
 
        '''
 
        """
 
        releases the pid by removing the pidfile
 
        '''
 
        """
 
        if self.callbackfn:
 
            #execute callback function on release
 
            if self.debug:
 
                print 'executing callback function %s' % self.callbackfn
 
            self.callbackfn()
 
        try:
 
            if self.debug:
 
                print 'removing pidfile %s' % self.pidfile
 
            os.remove(self.pidfile)
 
            self.held = False
 
        except OSError, e:
 
            if self.debug:
 
                print 'removing pidfile failed %s' % e
 
            pass
 

	
 
    def makelock(self, lockname, pidfile):
 
        '''
 
        """
 
        this function will make an actual lock
 
        @param lockname: acctual pid of file
 
        @param pidfile: the file to write the pid in
 
        '''
 
        """
 
        if self.debug:
 
            print 'creating a file %s and pid: %s' % (pidfile, lockname)
 
        pidfile = open(self.pidfile, "wb")
 
        pidfile.write(lockname)
 
        pidfile.close
 
        self.held = True
 

	
 

	
 
def main():
 
    print 'func is running'
 
    cnt = 20
 
    while 1:
pylons_app/lib/utils.py
Show inline comments
 
@@ -365,27 +365,26 @@ class OrderedDict(dict, DictMixin):
 
    def __ne__(self, other):
 
        return not self == other
 

	
 

	
 
#===============================================================================
 
# TEST FUNCTIONS
 
#===============================================================================
 
def create_test_index(repo_location, full_index):
 
    """Makes default test index
 
    @param repo_location:
 
    @param full_index:
 
    """
 
    from pylons_app.lib.indexers import daemon
 
    from pylons_app.lib.indexers.daemon import WhooshIndexingDaemon
 
    from pylons_app.lib.indexers.pidlock import DaemonLock, LockHeld
 
    from pylons_app.lib.pidlock import DaemonLock, LockHeld
 
    from pylons_app.lib.indexers import IDX_LOCATION
 
    import shutil
 
    
 
    if os.path.exists(IDX_LOCATION):
 
        shutil.rmtree(IDX_LOCATION)
 
         
 
    try:
 
        l = DaemonLock()
 
        WhooshIndexingDaemon(repo_location=repo_location)\
 
            .run(full_index=full_index)
 
        l.release()
 
    except LockHeld:
pylons_app/templates/files/files_browser.html
Show inline comments
 
@@ -20,44 +20,45 @@
 
	<div class="browser-body">
 
		<table class="code-browser">
 
		         <thead>
 
		             <tr>
 
		                 <th>${_('Name')}</th>
 
		                 <th>${_('Size')}</th>
 
		                 <th>${_('Mimetype')}</th>
 
		                 <th>${_('Revision')}</th>
 
		                 <th>${_('Last modified')}</th>
 
		                 <th>${_('Last commiter')}</th>
 
		             </tr>
 
		         </thead>
 

	
 
          		% if c.files_list.parent:
 
		         	<tr class="parity0">
 
		          		<td>
 
		          		% if c.files_list.parent:
 
		          			${h.link_to('..',h.url('files_home',repo_name=c.repo_name,revision=c.cur_rev,f_path=c.files_list.parent.path),class_="browser-dir")}
 
		          		%endif
 
		          		</td>
 
		          		<td></td>
 
		          		<td></td>
 
		          		<td></td>
 
		          		<td></td>
 
	          		<td></td>
 
		         	</tr>
 
          		%endif
 
		         	
 
		    %for cnt,node in enumerate(c.files_list,1):
 
				<tr class="parity${cnt%2}">
 
		             <td>
 
						${h.link_to(node.name,h.url('files_home',repo_name=c.repo_name,revision=c.cur_rev,f_path=node.path),class_=file_class(node))}
 
		             </td>
 
		             <td>
 
		                %if node.is_file():
 
		             		${h.format_byte_size(node.size,binary=True)}
 
		             	%endif
 
		             </td>
 
		             <td>
 
		              %if node.is_file():
 
		                  ${node.mimetype}
 
		              %endif
 
		             </td>
 
		             <td>
 
		             	%if node.is_file():
 
		             		${node.last_changeset.revision}
 
		             	%endif
 
		             </td>
 
		             <td>
0 comments (0 inline, 0 general)