def index_chunk_task(index, batch_id, rec_id, chunk):
    """Index a chunk of things.

    :arg index: the name of the index to index to
    :arg batch_id: the name for the batch this chunk belongs to
    :arg rec_id: the id for the record for this task
    :arg chunk: a (class, id_list) of things to index
    """
    cls, id_list = chunk

    try:
        # Pin to master db to avoid replication lag issues and stale
        # data.
        pin_this_thread()

        # Update record data.
        rec = Record.objects.get(pk=rec_id)
        rec.start_time = datetime.datetime.now()
        rec.message = u'Reindexing into %s' % index
        rec.status = Record.STATUS_IN_PROGRESS
        rec.save()

        index_chunk(cls, id_list, reraise=True)

        rec.mark_success()

    except Exception:
        rec.mark_fail(u'Errored out %s %s' % (
                sys.exc_type, sys.exc_value))
        raise

    finally:
        unpin_this_thread()

def _rebuild_kb_chunk(data):
    """Re-render a chunk of documents.

    Note: Don't use host components when making redirects to wiki pages; those
    redirects won't be auto-pruned when they're 404s.

    """
    log.info('Rebuilding %s documents.' % len(data))

    pin_this_thread()  # Stick to master.

    messages = []
    start = time.time()
    for pk in data:
        message = None
        try:
            document = Document.objects.get(pk=pk)

            # If we know a redirect link to be broken (i.e. if it looks like a
            # link to a document but the document isn't there), log an error:
            url = document.redirect_url()
            if (url and points_to_document_view(url) and
                    not document.redirect_document()):
                log.warn('Invalid redirect document: %d' % pk)

            html = document.parse_and_calculate_links()
            if document.html != html:
                # We are calling update here to so we only update the html
                # column instead of all of them. This bypasses post_save
                # signal handlers like the one that triggers reindexing.
                # See bug 797038 and bug 797352.
                Document.objects.filter(pk=pk).update(html=html)
                statsd.incr('wiki.rebuild_chunk.change')
            else:
                statsd.incr('wiki.rebuild_chunk.nochange')
        except Document.DoesNotExist:
            message = 'Missing document: %d' % pk
        except Revision.DoesNotExist:
            message = 'Missing revision for document: %d' % pk
        except ValidationError as e:
            message = 'ValidationError for %d: %s' % (pk, e.messages[0])
        except SlugCollision:
            message = 'SlugCollision: %d' % pk
        except TitleCollision:
            message = 'TitleCollision: %d' % pk

        if message:
            log.debug(message)
            messages.append(message)
    d = time.time() - start
    statsd.timing('wiki.rebuild_chunk', int(round(d * 1000)))

    if messages:
        subject = ('[%s] Exceptions raised in _rebuild_kb_chunk()' %
                   settings.PLATFORM_NAME)
        mail_admins(subject=subject, message='\n'.join(messages))
    if not transaction.get_connection().in_atomic_block:
        transaction.commit()

    unpin_this_thread()  # Not all tasks need to do use the master.

def index_task(cls, id_list, **kw):
    """Index documents specified by cls and ids"""
    statsd.incr('search.tasks.index_task.%s' % cls.get_mapping_type_name())
    try:
        # Pin to master db to avoid replication lag issues and stale
        # data.
        pin_this_thread()

        qs = cls.get_model().uncached.filter(id__in=id_list).values_list(
            'id', flat=True)
        for id_ in qs:
            try:
                cls.index(cls.extract_document(id_), id_=id_)
            except UnindexMeBro:
                # If extract_document throws this, then we need to
                # remove this item from the index.
                cls.unindex(id_)

    except Exception as exc:
        retries = index_task.request.retries
        if retries >= MAX_RETRIES:
            # Some exceptions aren't pickleable and we need this to
            # throw things that are pickleable.
            raise IndexingTaskError()

        statsd.incr('search.tasks.index_task.retry', 1)
        statsd.incr('search.tasks.index_task.retry%d' % RETRY_TIMES[retries],
                    1)

        index_task.retry(exc=exc, max_retries=MAX_RETRIES,
                         countdown=RETRY_TIMES[retries])
    finally:
        unpin_this_thread()

 def test_pinning_encapsulation(self):
     """Check the pinning getters and setters."""
     self.assertFalse(this_thread_is_pinned())
     pin_this_thread()
     self.assertTrue(this_thread_is_pinned())
     unpin_this_thread()
     self.assertFalse(this_thread_is_pinned())

def log_answer(answer):
    pin_this_thread()

    creator = answer.creator
    created = answer.created
    question = answer.question
    users = [a.creator for a in
             question.answers.select_related('creator').exclude(
                creator=creator)]
    if question.creator != creator:
        users += [question.creator]
    users = set(users)  # Remove duplicates.

    if users:
        action = Action.objects.create(
            creator=creator,
            created=created,
            url=answer.get_absolute_url(),
            content_object=answer,
            formatter_cls='questions.formatters.AnswerFormatter')
        action.users.add(*users)

    transaction.commit_unless_managed()

    # Record karma actions
    AnswerAction(answer.creator, answer.created.date()).save()
    if answer == answer.question.answers.order_by('created')[0]:
        FirstAnswerAction(answer.creator, answer.created.date()).save()

    unpin_this_thread()

def _rebuild_kb_chunk(data, **kwargs):
    """Re-render a chunk of documents."""
    log.info('Rebuilding %s documents.' % len(data))

    pin_this_thread()  # Stick to master.

    messages = []
    for pk in data:
        message = None
        try:
            document = Document.objects.get(pk=pk)
            document.html = document.current_revision.content_cleaned
            document.save()
        except Document.DoesNotExist:
            message = 'Missing document: %d' % pk
        except ValidationError as e:
            message = 'ValidationError for %d: %s' % (pk, e.messages[0])
        except SlugCollision:
            message = 'SlugCollision: %d' % pk
        except TitleCollision:
            message = 'TitleCollision: %d' % pk

        if message:
            log.debug(message)
            messages.append(message)

    if messages:
        subject = ('[%s] Exceptions raised in _rebuild_kb_chunk()' %
                   settings.PLATFORM_NAME)
        mail_admins(subject=subject, message='\n'.join(messages))
    transaction.commit_unless_managed()

    unpin_this_thread()  # Not all tasks need to do use the master.

def render_document_cascade(base):
    """Given a document, render it and all documents that may be affected."""

    # This walks along the graph of links between documents. If there is
    # a document A that includes another document B as a template, then
    # there is an edge from A to B in this graph. The goal here is to
    # process every node exactly once. This is robust to cycles and
    # diamonds in the graph, since it keeps track of what nodes have
    # been visited already.

    # In case any thing goes wrong, this guarantees we unpin the DB
    try:
        # Sends all writes to the master DB. Slaves are readonly.
        pin_this_thread()

        todo = set([base])
        done = set()

        while todo:
            d = todo.pop()
            if d in done:
                # Don't process a node twice.
                continue
            d.html = d.parse_and_calculate_links()
            d.save()
            done.add(d)
            todo.update(l.linked_from for l in d.links_to()
                        .filter(kind__in=['template', 'include']))

    finally:
        unpin_this_thread()

 def test_decorator_resets(self):
     @use_master
     def check():
         assert this_thread_is_pinned()
     pin_this_thread()
     assert this_thread_is_pinned()
     check()
     assert this_thread_is_pinned()

 def test_decorator_resets(self):
     @use_primary_db
     def check():
         assert this_thread_is_pinned()
     pin_this_thread()
     assert this_thread_is_pinned()
     check()
     assert this_thread_is_pinned()

 def process_request(self, request):
     """Set the thread's pinning flag according to the presence of the
     incoming cookie."""
     if PINNING_COOKIE in request.COOKIES or request.method not in READ_ONLY_METHODS:
         pin_this_thread()
     else:
         # In case the last request this thread served was pinned:
         unpin_this_thread()

 def test_slave_context_manager_exception(self):
     pin_this_thread()
     self.assertTrue(this_thread_is_pinned())
     with self.assertRaises(ValueError):
         with use_slave:
             self.assertFalse(this_thread_is_pinned())
             raise ValueError
     self.assertTrue(this_thread_is_pinned())

        def thread2_worker():
            pin_this_thread()
            with use_primary_db:
                orchestrator.release()
                thread2_lock.acquire()

            pinned[2] = this_thread_is_pinned()
            orchestrator.release()

    def test_pinned_reads(self):
        """Test PinningMasterSlaveRouter.db_for_read() when pinned and when
        not."""
        router = PinningMasterSlaveRouter()

        eq_(router.db_for_read(TestModel), get_slave())

        pin_this_thread()
        eq_(router.db_for_read(TestModel), MASTER_DATABASE)

    def test_pinned_reads(self):
        """Test PinningMasterSlaveRouter.db_for_read() when pinned and when
        not."""
        router = PinningMasterSlaveRouter()

        eq_(router.db_for_read(None), get_slave())

        pin_this_thread()
        eq_(router.db_for_read(None), DEFAULT_DB_ALIAS)

    def test_pinning_encapsulation(self):
        """Check the pinning getters and setters."""
        assert not this_thread_is_pinned(), "Thread started out pinned or this_thread_is_pinned() is broken."

        pin_this_thread()
        assert this_thread_is_pinned(), "pin_this_thread() didn't pin the thread."

        unpin_this_thread()
        assert not this_thread_is_pinned(), "Thread remained pinned after unpin_this_thread()."

    def test_decorator_resets(self):

        @use_master
        def check():
            self.assertTrue(this_thread_is_pinned())

        pin_this_thread()
        self.assertTrue(this_thread_is_pinned())
        check()
        self.assertTrue(this_thread_is_pinned())

    def test_slave_decorator(self):

        @use_slave
        def check():
            self.assertFalse(this_thread_is_pinned())

        pin_this_thread()
        self.assertTrue(this_thread_is_pinned())
        check()
        self.assertTrue(this_thread_is_pinned())

    def process_request(self, request):
        if not getattr(request, 'API', False):
            return super(APIPinningMiddleware, self).process_request(request)

        if request.amo_user and not request.amo_user.is_anonymous():
            statsd.incr('api.db.pinned')
            pin_this_thread()
            return

        statsd.incr('api.db.unpinned')
        unpin_this_thread()

    def process_request(self, request):
        if not getattr(request, 'API', False):
            return super(APIPinningMiddleware, self).process_request(request)

        if (request.user and not request.user.is_anonymous() and
                (cache.get(self.cache_key(request)) or
                 request.method in ['DELETE', 'PATCH', 'POST', 'PUT'])):
            statsd.incr('api.db.pinned')
            pin_this_thread()
            return

        statsd.incr('api.db.unpinned')
        unpin_this_thread()

def migrate_helpfulvotes(start_id, end_id):
    """Transfer helpfulvotes from old to new version."""

    if not waffle.switch_is_active('migrate-helpfulvotes'):
        raise  # Celery emails the failed IDs so we know to which to rerun.

    start = time.time()

    pin_this_thread()  # Pin to master

    transaction.enter_transaction_management()
    transaction.managed(True)
    try:
        cursor = connection.cursor()
        cursor.execute("""INSERT INTO `wiki_helpfulvote`
            (revision_id, helpful, created,
            creator_id, anonymous_id, user_agent)
            SELECT COALESCE(
                    (SELECT id FROM `wiki_revision`
                        WHERE `document_id` = wiki_helpfulvoteold.document_id
                            AND `is_approved`=1 AND
                            (`reviewed` <= wiki_helpfulvoteold.created
                                OR `reviewed` IS NULL)
                        ORDER BY CASE WHEN `reviewed`
                            IS NULL THEN 1 ELSE 0 END,
                                  `wiki_revision`.`created` DESC LIMIT 1),
                    (SELECT id FROM `wiki_revision`
                        WHERE `document_id` = wiki_helpfulvoteold.document_id
                            AND (`reviewed` <= wiki_helpfulvoteold.created
                                OR `reviewed` IS NULL)
                        ORDER BY CASE WHEN `reviewed`
                            IS NULL THEN 1 ELSE 0 END,
                                `wiki_revision`.`created`  DESC LIMIT 1),
                    (SELECT id FROM `wiki_revision`
                        WHERE `document_id` = wiki_helpfulvoteold.document_id
                        ORDER BY `created` ASC LIMIT 1)),
                helpful, created, creator_id, anonymous_id, user_agent
            FROM `wiki_helpfulvoteold` WHERE id >= %s AND id < %s""",
            [start_id, end_id])
        transaction.commit()
    except:
        transaction.rollback()
        raise

    transaction.leave_transaction_management()

    unpin_this_thread()

    d = time.time() - start
    statsd.timing('wiki.migrate_helpfulvotes', int(round(d * 1000)))