kallithea Changeset - 0c43c6671815

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

rhodecode/lib/celerylib/tasks.py

0 comments (0 inline, 0 general)

rhodecode/lib/celerylib/__init__.py

➞

Show inline comments

@@ @@ -48,6 +48,7 @@ try: @@
 except KeyError:
     CELERY_ON = False
 class ResultWrapper(object):
     def __init__(self, task):
         self.task = task
@@ @@ -56,12 +57,14 @@ class ResultWrapper(object): @@
     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')
@@ @@ -76,14 +79,20 @@ def run_task(task, *args, **kwargs): @@
     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)

rhodecode/lib/celerylib/tasks.py

➞

Show inline comments

@@ @@ -37,13 +37,14 @@ 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
@@ @@ -125,16 +126,19 @@ def whoosh_index(repo_location, full_ind @@
 @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 = {}
@@ @@ -142,6 +146,11 @@ def get_commits_stats(repo_name, ts_min_ @@
     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'))
@@ @@ -159,18 +168,15 @@ def get_commits_stats(repo_name, ts_min_ @@
     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)
@@ @@ -194,7 +200,8 @@ def get_commits_stats(repo_name, ts_min_ @@
             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)
@@ @@ -232,7 +239,9 @@ def get_commits_stats(repo_name, ts_min_ @@
         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),
@@ @@ -260,11 +269,19 @@ def get_commits_stats(repo_name, ts_min_ @@
     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)
@@ @@ -313,7 +330,6 @@ 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
@@ @@ -351,14 +367,14 @@ def send_email(recipients, subject, body @@
 @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']

0 comments (0 inline, 0 general)