From c0d9fde4bf1ba6d50b846fff5b71b39687832345 Mon Sep 17 00:00:00 2001 From: Clement Verna Date: Feb 21 2017 20:36:28 +0000 Subject: [PATCH 1/2] Added multi thread to fedorahosted issue import Signed-off-by: Clement Verna --- diff --git a/pagure_importer/commands/fedorahosted.py b/pagure_importer/commands/fedorahosted.py index cc4d822..44412a5 100644 --- a/pagure_importer/commands/fedorahosted.py +++ b/pagure_importer/commands/fedorahosted.py @@ -1,5 +1,7 @@ import os from collections import namedtuple +from threading import Thread +from queue import Queue import click import pagure_importer from pagure_importer.app import app, REPO_PATH @@ -40,8 +42,8 @@ def fedorahosted(project_url, tags, private, username, password, project = get_pagure_namespace(REPO_PATH, repo_name) rest_param = namedtuple('rest_param', 'url username password') rest = rest_param(project_url, username, password) - fas = FASclient(rest.username, rest.password, - 'https://admin.fedoraproject.org/accounts') + + queue = Queue() # import issues tickets_id = libtrac.get_project_tickets(rest=rest) @@ -50,34 +52,51 @@ def fedorahosted(project_url, tags, private, username, password, clone_repo_location = os.path.join(REPO_PATH, 'clone-' + repo_name) for ticket_id in tickets_id: + queue.put(ticket_id) - pagure_issue = libtrac.create_issue(ticket_id=ticket_id, - custom_fields=custom_fields, - rest=rest, - fas=fas, - offset=offset, - private=private, - tags=tags) - pagure_issue.comments = [] - pagure_issue_comments = libtrac.get_ticket_comments(rest=rest, - ticket_id=ticket_id) - comments = libtrac.create_comments(trac_comments=pagure_issue_comments, - fas=fas) - # add all the comments to the issue object - pagure_issue = libtrac.add_comment_to_issue(comments=comments, - pagure_issue=pagure_issue, - repo_name=project) + for i in range(30): + t = Thread(target=import_issues, + args=(rest, queue, custom_fields, offset, private, tags, project, + clone_repo_location, tickets_id), + daemon=True) + t.start() - click.echo('Updated ' + repo_name + ' with issue :' + - str(ticket_id) + '/' + str(tickets_id[-1])) - issue_to_json(issue=pagure_issue, folder=clone_repo_location) + queue.join() - # update the local git repo - new_repo = gitutils.update_git( - new_repo, - commit_message='Imported issues from fedorahosted project: %s' % repo_name) + # update the local git repo + new_repo = gitutils.update_git( + new_repo, + commit_message='Imported issues from fedorahosted project: %s' % repo_name) - if not nopush: - gitutils.push_repo(new_repo) + if not nopush: + gitutils.push_repo(new_repo) else: click.echo('No ticket repository found. Use pgimport clone command') + +def import_issues(rest, queue, custom_fields, offset, private, tags, project, + clone_repo_location, tickets_id): + + fas = FASclient(rest.username, rest.password, + 'https://admin.fedoraproject.org/accounts') + while True: + ticket_id = queue.get() + pagure_issue = libtrac.create_issue(ticket_id=ticket_id, + custom_fields=custom_fields, + rest=rest, + fas=fas, + offset=offset, + private=private, + tags=tags) + pagure_issue.comments = [] + pagure_issue_comments = libtrac.get_ticket_comments(rest=rest, + ticket_id=ticket_id) + comments = libtrac.create_comments(trac_comments=pagure_issue_comments, + fas=fas) + # add all the comments to the issue object + pagure_issue = libtrac.add_comment_to_issue(comments=comments, + pagure_issue=pagure_issue, + repo_name=project) + + click.echo('Updating issue : {}/{}'.format(str(ticket_id), str(tickets_id[-1]))) + issue_to_json(issue=pagure_issue, folder=clone_repo_location) + queue.task_done() From 71397f0dddbda7f108118b1f79cb67c8c11821ee Mon Sep 17 00:00:00 2001 From: Clement Verna Date: Feb 22 2017 21:50:10 +0000 Subject: [PATCH 2/2] Adding initial support for multithreading. When a thread dies it will kill the main thread which will then kill all the other running threads Signed-off-by: Clement Verna --- diff --git a/pagure_importer/commands/fedorahosted.py b/pagure_importer/commands/fedorahosted.py index 44412a5..e09ec46 100644 --- a/pagure_importer/commands/fedorahosted.py +++ b/pagure_importer/commands/fedorahosted.py @@ -1,4 +1,7 @@ import os +import threading +import signal +import traceback from collections import namedtuple from threading import Thread from queue import Queue @@ -25,8 +28,10 @@ import pagure_importer.utils.importer_trac as libtrac help='Number of issue in pagure before import') @click.option('--nopush', is_flag=True, help="Do not push the result of pagure-importer back") +@click.option('--threads', default=1, + help='Number of thread running the import') def fedorahosted(project_url, tags, private, username, password, - offset, nopush): + offset, nopush, threads): """Import issues from fedorahosted""" if project_url.endswith('.git'): @@ -51,14 +56,15 @@ def fedorahosted(project_url, tags, private, username, password, custom_fields = libtrac.get_custom_fields(ticket_fields=ticket_fields) clone_repo_location = os.path.join(REPO_PATH, 'clone-' + repo_name) + signal.signal(signal.SIGQUIT, signal.SIG_DFL) for ticket_id in tickets_id: queue.put(ticket_id) - for i in range(30): + for i in range(threads): t = Thread(target=import_issues, args=(rest, queue, custom_fields, offset, private, tags, project, - clone_repo_location, tickets_id), - daemon=True) + clone_repo_location, tickets_id)) + t.setDaemon(True) t.start() queue.join() @@ -73,30 +79,37 @@ def fedorahosted(project_url, tags, private, username, password, else: click.echo('No ticket repository found. Use pgimport clone command') + def import_issues(rest, queue, custom_fields, offset, private, tags, project, clone_repo_location, tickets_id): fas = FASclient(rest.username, rest.password, 'https://admin.fedoraproject.org/accounts') while True: - ticket_id = queue.get() - pagure_issue = libtrac.create_issue(ticket_id=ticket_id, - custom_fields=custom_fields, - rest=rest, - fas=fas, - offset=offset, - private=private, - tags=tags) - pagure_issue.comments = [] - pagure_issue_comments = libtrac.get_ticket_comments(rest=rest, - ticket_id=ticket_id) - comments = libtrac.create_comments(trac_comments=pagure_issue_comments, - fas=fas) - # add all the comments to the issue object - pagure_issue = libtrac.add_comment_to_issue(comments=comments, - pagure_issue=pagure_issue, - repo_name=project) - - click.echo('Updating issue : {}/{}'.format(str(ticket_id), str(tickets_id[-1]))) - issue_to_json(issue=pagure_issue, folder=clone_repo_location) - queue.task_done() + try: + ticket_id = queue.get() + pagure_issue = libtrac.create_issue(ticket_id=ticket_id, + custom_fields=custom_fields, + rest=rest, + fas=fas, + offset=offset, + private=private, + tags=tags) + pagure_issue.comments = [] + pagure_issue_comments = libtrac.get_ticket_comments(rest=rest, + ticket_id=ticket_id) + comments = libtrac.create_comments(trac_comments=pagure_issue_comments, + fas=fas) + # add all the comments to the issue object + pagure_issue = libtrac.add_comment_to_issue(comments=comments, + pagure_issue=pagure_issue, + repo_name=project) + + click.echo('Updating issue : {}/{}'.format(str(ticket_id), str(tickets_id[-1]))) + issue_to_json(issue=pagure_issue, folder=clone_repo_location) + queue.task_done() + except: + click.echo('Exception while importing issue {}'.format(ticket_id)) + traceback.print_exc() + main = threading.main_thread() + signal.pthread_kill(main.ident, signal.SIGQUIT)