Changeset - 0c43c6671815
Parent rev.
Child rev.
[Not reviewed]
beta
0 2 0
Marcin Kuzminski - 15 years ago 2011-04-17 14:45:16
marcin@python-works.com
moved locking of commit stats into the task itself to remove race conditions when lock was not removed before starting another task.
2 files changed with 50 insertions and 25 deletions:
rhodecode/lib/celerylib/__init__.py
13
4
rhodecode/lib/celerylib/tasks.py
37
21
0 comments (0 inline, 0 general)
rhodecode/lib/celerylib/__init__.py
➞
Show inline comments
 
@@ -45,26 +45,29 @@ log = logging.getLogger(__name__)
 

			




	
	
	
		
 
try:

			




	
	
	
		
 
    CELERY_ON = str2bool(config['app_conf'].get('use_celery'))

			




	
	
	
		
 
except KeyError:

			




	
	
	
		
 
    CELERY_ON = False

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
class ResultWrapper(object):

			




	
	
	
		
 
    def __init__(self, task):

			




	
	
	
		
 
        self.task = task

			




	
	
	
		
 

			




	
	
	
		
 
    @LazyProperty

			




	
	
	
		
 
    def result(self):

			




	
	
	
		
 
        return self.task

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def run_task(task, *args, **kwargs):

			




	
	
	
		
 
    if CELERY_ON:

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            t = task.apply_async(args=args, kwargs=kwargs)

			




	
	
	
		
 
            log.info('running task %s:%s', t.task_id, task)

			




	
	
	
		
 
            return t

			




	
	
	
		
 

			




	
	
	
		
 
        except socket.error, e:

			




	
	
	
		
 
            if  e.errno == 111:

			




	
	
	
		
 
                log.debug('Unable to connect to celeryd. Sync execution')

			




	
	
	
		
 
            else:

			




	
	
	
		
 
                log.error(traceback.format_exc())

			




	
	
	
		
 
        except KeyError, e:

			




	
	
		
 
@@ -73,20 +76,26 @@ def run_task(task, *args, **kwargs):

			




	
	
	
		
 
            log.error(traceback.format_exc())

			




	
	
	
		
 

			




	
	
	
		
 
    log.debug('executing task %s in sync mode', task)

			




	
	
	
		
 
    return ResultWrapper(task(*args, **kwargs))

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def locked_task(func):

			




	
	
	
		
 
    def __wrapper(func, *fargs, **fkwargs):

			




	
	
	
		
 
def __get_lockkey(func, *fargs, **fkwargs):

			




	
	
	
		
 
        params = list(fargs)

			




	
	
	
		
 
        params.extend(['%s-%s' % ar for ar in fkwargs.items()])

			




	
	
	
		
 

			




	
	
	
		
 
    func_name = str(func.__name__) if hasattr(func, '__name__') else str(func)

			




	
	
	
		
 

			




	
	
	
		
 
        lockkey = 'task_%s' % \

			




	
	
	
		
 
            md5(str(func.__name__) + '-' + \

			




	
	
	
		
 
                '-'.join(map(str, params))).hexdigest()

			




	
	
	
		
 
        md5(func_name + '-' + '-'.join(map(str, params))).hexdigest()

			




	
	
	
		
 
    return lockkey

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
def locked_task(func):

			




	
	
	
		
 
    def __wrapper(func, *fargs, **fkwargs):

			




	
	
	
		
 
        lockkey = __get_lockkey(func, *fargs, **fkwargs)

			




	
	
	
		
 
        log.info('running task with lockkey %s', lockkey)

			




	
	
	
		
 
        try:

			




	
	
	
		
 
            l = DaemonLock(lockkey)

			




	
	
	
		
 
            ret = func(*fargs, **fkwargs)

			




	
	
	
		
 
            l.release()

			




	
	
	
		
 
            return ret

			




        

    


    
    

    

        

                

                    rhodecode/lib/celerylib/tasks.py
                

                

                  
                      
                        

                      ➞

                      
                        
                  

                  
                      
                  
                      
                  
                      
                  
                      
                  
                  
                

                

                    Show inline comments
                    
                

        

        

            



	
	
		
 
@@ -34,19 +34,20 @@ from operator import itemgetter

			




	
	
	
		
 
from pygments import lexers

			




	
	
	
		
 
from string import lower

			




	
	
	
		
 

			




	
	
	
		
 
from pylons import config

			




	
	
	
		
 
from pylons.i18n.translation import _

			




	
	
	
		
 

			




	
	
	
		
 
from rhodecode.lib.celerylib import run_task, locked_task, str2bool

			




	
	
	
		
 
from rhodecode.lib.celerylib import run_task, locked_task, str2bool, \

			




	
	
	
		
 
    __get_lockkey, LockHeld, DaemonLock

			




	
	
	
		
 
from rhodecode.lib.helpers import person

			




	
	
	
		
 
from rhodecode.lib.smtp_mailer import SmtpMailer

			




	
	
	
		
 
from rhodecode.lib.utils import OrderedDict, add_cache

			




	
	
	
		
 
from rhodecode.model import init_model

			




	
	
	
		
 
from rhodecode.model import meta

			




	
	
	
		
 
from rhodecode.model.db import RhodeCodeUi

			




	
	
	
		
 
from rhodecode.model.db import RhodeCodeUi, Statistics, Repository

			




	
	
	
		
 

			




	
	
	
		
 
from vcs.backends import get_repo

			




	
	
	
		
 

			




	
	
	
		
 
from sqlalchemy import engine_from_config

			




	
	
	
		
 

			




	
	
	
		
 
add_cache(config)

			




	
	
		
 
@@ -122,29 +123,37 @@ def whoosh_index(repo_location, full_ind

			




	
	
	
		
 
    WhooshIndexingDaemon(index_location=index_location,

			




	
	
	
		
 
                         repo_location=repo_location, sa=get_session())\

			




	
	
	
		
 
                         .run(full_index=full_index)

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
@task(ignore_result=True)

			




	
	
	
		
 
@locked_task

			




	
	
	
		
 
def get_commits_stats(repo_name, ts_min_y, ts_max_y):

			




	
	
	
		
 
    try:

			




	
	
	
		
 
        log = get_commits_stats.get_logger()

			




	
	
	
		
 
    except:

			




	
	
	
		
 
        log = logging.getLogger(__name__)

			




	
	
	
		
 

			




	
	
	
		
 
    from rhodecode.model.db import Statistics, Repository

			




	
	
	
		
 
    lockkey = __get_lockkey('get_commits_stats', repo_name, ts_min_y,

			




	
	
	
		
 
                            ts_max_y)

			




	
	
	
		
 
    log.info('running task with lockkey %s', lockkey)

			




	
	
	
		
 
    try:

			




	
	
	
		
 
        lock = DaemonLock(lockkey)

			




	
	
	
		
 

			




	
	
	
		
 
    #for js data compatibilty

			




	
	
	
		
 
        #for js data compatibilty cleans the key for person from '

			




	
	
	
		
 
    akc = lambda k: person(k).replace('"', "")

			




	
	
	
		
 

			




	
	
	
		
 
    co_day_auth_aggr = {}

			




	
	
	
		
 
    commits_by_day_aggregate = {}

			




	
	
	
		
 
    repos_path = get_repos_path()

			




	
	
	
		
 
    p = os.path.join(repos_path, repo_name)

			




	
	
	
		
 
    repo = get_repo(p)

			




	
	
	
		
 
        repo_size = len(repo.revisions)

			




	
	
	
		
 
        #return if repo have no revisions

			




	
	
	
		
 
        if repo_size < 1:

			




	
	
	
		
 
            lock.release()

			




	
	
	
		
 
            return True

			




	
	
	
		
 

			




	
	
	
		
 
    skip_date_limit = True

			




	
	
	
		
 
    parse_limit = int(config['app_conf'].get('commit_parse_limit'))

			




	
	
	
		
 
    last_rev = 0

			




	
	
	
		
 
    last_cs = None

			




	
	
	
		
 
    timegetter = itemgetter('time')

			




	
	
		
 
@@ -156,24 +165,21 @@ def get_commits_stats(repo_name, ts_min_

			




	
	
	
		
 
    cur_stats = sa.query(Statistics)\

			




	
	
	
		
 
        .filter(Statistics.repository == dbrepo).scalar()

			




	
	
	
		
 

			




	
	
	
		
 
    if cur_stats is not None:

			




	
	
	
		
 
        last_rev = cur_stats.stat_on_revision

			




	
	
	
		
 

			




	
	
	
		
 
    #return if repo is empty

			




	
	
	
		
 
    if not repo.revisions:

			




	
	
	
		
 
        return True

			




	
	
	
		
 

			




	
	
	
		
 
    if last_rev == repo.get_changeset().revision and len(repo.revisions) > 1:

			




	
	
	
		
 
        if last_rev == repo.get_changeset().revision and repo_size > 1:

			




	
	
	
		
 
        #pass silently without any work if we're not on first revision or

			




	
	
	
		
 
        #current state of parsing revision(from db marker) is the last revision

			




	
	
	
		
 
            #current state of parsing revision(from db marker) is the

			




	
	
	
		
 
            #last revision

			




	
	
	
		
 
            lock.release()

			




	
	
	
		
 
        return True

			




	
	
	
		
 

			




	
	
	
		
 
    if cur_stats:

			




	
	
	
		
 
        commits_by_day_aggregate = OrderedDict(

			




	
	
	
		
 
                                       json.loads(

			




	
	
	
		
 
            commits_by_day_aggregate = OrderedDict(json.loads(

			




	
	
	
		
 
                                        cur_stats.commit_activity_combined))

			




	
	
	
		
 
        co_day_auth_aggr = json.loads(cur_stats.commit_activity)

			




	
	
	
		
 

			




	
	
	
		
 
    log.debug('starting parsing %s', parse_limit)

			




	
	
	
		
 
    lmktime = mktime

			




	
	
	
		
 

			




	
	
		
 
@@ -191,13 +197,14 @@ def get_commits_stats(repo_name, ts_min_

			




	
	
	
		
 
                time_pos = l.index(k)

			




	
	
	
		
 
            except ValueError:

			




	
	
	
		
 
                time_pos = False

			




	
	
	
		
 

			




	
	
	
		
 
            if time_pos >= 0 and time_pos is not False:

			




	
	
	
		
 

			




	
	
	
		
 
                datadict = co_day_auth_aggr[akc(cs.author)]['data'][time_pos]

			




	
	
	
		
 
                    datadict = \

			




	
	
	
		
 
                        co_day_auth_aggr[akc(cs.author)]['data'][time_pos]

			




	
	
	
		
 

			




	
	
	
		
 
                datadict["commits"] += 1

			




	
	
	
		
 
                datadict["added"] += len(cs.added)

			




	
	
	
		
 
                datadict["changed"] += len(cs.changed)

			




	
	
	
		
 
                datadict["removed"] += len(cs.removed)

			




	
	
	
		
 

			




	
	
		
 
@@ -229,13 +236,15 @@ def get_commits_stats(repo_name, ts_min_

			




	
	
	
		
 
        #gather all data by day

			




	
	
	
		
 
        if k in commits_by_day_aggregate:

			




	
	
	
		
 
            commits_by_day_aggregate[k] += 1

			




	
	
	
		
 
        else:

			




	
	
	
		
 
            commits_by_day_aggregate[k] = 1

			




	
	
	
		
 

			




	
	
	
		
 
    overview_data = sorted(commits_by_day_aggregate.items(), key=itemgetter(0))

			




	
	
	
		
 
        overview_data = sorted(commits_by_day_aggregate.items(),

			




	
	
	
		
 
                               key=itemgetter(0))

			




	
	
	
		
 

			




	
	
	
		
 
    if not co_day_auth_aggr:

			




	
	
	
		
 
        co_day_auth_aggr[akc(repo.contact)] = {

			




	
	
	
		
 
            "label": akc(repo.contact),

			




	
	
	
		
 
            "data": [0, 1],

			




	
	
	
		
 
            "schema": ["commits"],

			




	
	
	
		
 
        }

			




	
	
		
 
@@ -257,17 +266,25 @@ def get_commits_stats(repo_name, ts_min_

			




	
	
	
		
 
        stats.stat_on_revision = last_cs.revision if last_cs else 0

			




	
	
	
		
 
        sa.add(stats)

			




	
	
	
		
 
        sa.commit()

			




	
	
	
		
 
    except:

			




	
	
	
		
 
        log.error(traceback.format_exc())

			




	
	
	
		
 
        sa.rollback()

			




	
	
	
		
 
            lock.release()

			




	
	
	
		
 
        return False

			




	
	
	
		
 
    if len(repo.revisions) > 1:

			




	
	
	
		
 

			




	
	
	
		
 
        #final release

			




	
	
	
		
 
        lock.release()

			




	
	
	
		
 

			




	
	
	
		
 
        #execute another task if celery is enabled

			




	
	
	
		
 
        if len(repo.revisions) > 1 and CELERY_ON:

			




	
	
	
		
 
        run_task(get_commits_stats, repo_name, ts_min_y, ts_max_y)

			




	
	
	
		
 

			




	
	
	
		
 
    return True

			




	
	
	
		
 
    except LockHeld:

			




	
	
	
		
 
        log.info('LockHeld')

			




	
	
	
		
 
        return 'Task with key %s already running' % lockkey

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
@task(ignore_result=True)

			




	
	
	
		
 
def reset_user_password(user_email):

			




	
	
	
		
 
    try:

			




	
	
	
		
 
        log = reset_user_password.get_logger()

			




	
	
		
 
@@ -310,13 +327,12 @@ def reset_user_password(user_email):

			




	
	
	
		
 

			




	
	
	
		
 
@task(ignore_result=True)

			




	
	
	
		
 
def send_email(recipients, subject, body):

			




	
	
	
		
 
    """

			




	
	
	
		
 
    Sends an email with defined parameters from the .ini files.

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
    :param recipients: list of recipients, it this is empty the defined email

			




	
	
	
		
 
        address from field 'email_to' is used instead

			




	
	
	
		
 
    :param subject: subject of the mail

			




	
	
	
		
 
    :param body: body of the mail

			




	
	
	
		
 
    """

			




	
	
	
		
 
    try:

			




	
	
		
 
@@ -348,20 +364,20 @@ def send_email(recipients, subject, body

			




	
	
	
		
 
        return False

			




	
	
	
		
 
    return True

			




	
	
	
		
 

			




	
	
	
		
 

			




	
	
	
		
 
@task(ignore_result=True)

			




	
	
	
		
 
def create_repo_fork(form_data, cur_user):

			




	
	
	
		
 
    from rhodecode.model.repo import RepoModel

			




	
	
	
		
 
    from vcs import get_backend

			




	
	
	
		
 

			




	
	
	
		
 
    try:

			




	
	
	
		
 
        log = create_repo_fork.get_logger()

			




	
	
	
		
 
    except:

			




	
	
	
		
 
        log = logging.getLogger(__name__)

			




	
	
	
		
 

			




	
	
	
		
 
    from rhodecode.model.repo import RepoModel

			




	
	
	
		
 
    from vcs import get_backend

			




	
	
	
		
 

			




	
	
	
		
 
    repo_model = RepoModel(get_session())

			




	
	
	
		
 
    repo_model.create(form_data, cur_user, just_db=True, fork=True)

			




	
	
	
		
 
    repo_name = form_data['repo_name']

			




	
	
	
		
 
    repos_path = get_repos_path()

			




	
	
	
		
 
    repo_path = os.path.join(repos_path, repo_name)

			




	
	
	
		
 
    repo_fork_path = os.path.join(repos_path, form_data['fork_name'])

			




        

    



    


    



    



      

      





    
    0 comments (0 inline, 0 general)
    





    


  

  







  


    




    








    
        
    
    
        This site is powered by
            Kallithea 0.7.0,
        which is
        © 2010–2025 by various authors & licensed under GPLv3.