<commit_before># Copyright (c) 2014 Rackspace, Inc.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import division
from __future__ import print_function

import multiprocessing as mp
import sys
import time

from gevent import monkey as curious_george
curious_george.patch_all(thread=False, select=False)
import gevent
import marktime
from zaqarclient.queues.v1 import client
from zaqarclient.transport.errors import TransportError

from zaqar.bench.config import conf


def claim_delete(stats, test_duration, ttl, grace, limit):
    """Consumer Worker

    The Consumer Worker continuously claims and deletes messages
    for the specified duration. The time taken for each claim and
    delete is recorded for calculating throughput and latency.
    """

    cli = client.Client(conf.server_url)
    queue = cli.queue(conf.queue_prefix + '1')
    end = time.time() + test_duration
    claim_total_elapsed = 0
    delete_total_elapsed = 0
    total_requests = 0
    claim_total_requests = 0
    delete_total_requests = 0

    while time.time() < end:
        marktime.start('claim_message')
        try:
            claim = queue.claim(ttl=ttl, grace=grace, limit=limit)

        except TransportError as ex:
            sys.stderr.write("Could not claim messages : {0}\n".format(ex))

        else:
            claim_total_elapsed += marktime.stop('claim_message').seconds
            claim_total_requests += 1

            try:
                marktime.start('delete_message')

                for msg in claim:
                    # TODO(TheSriram): Simulate actual work before deletion
                    delete_total_requests += 1
                    msg.delete()

                delete_total_elapsed += marktime.stop('delete_message').seconds

            except TransportError as ex:
                sys.stderr.write("Could not delete messages: {0}\n".format(ex))

            finally:
                total_requests += 1
        finally:
            total_requests += 1

    stats.put({
        'total_requests': total_requests,
        'claim_total_requests': claim_total_requests,
        'delete_total_requests': delete_total_requests,
        'claim_total_elapsed': claim_total_elapsed,
        'delete_total_elapsed': delete_total_elapsed,
    })

<commit_msg>Fix incorrect accounting of total requests in benchmark consumer<commit_after># Copyright (c) 2014 Rackspace, Inc.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from __future__ import division
from __future__ import print_function

import multiprocessing as mp
import sys
import time

from gevent import monkey as curious_george
curious_george.patch_all(thread=False, select=False)
import gevent
import marktime
from zaqarclient.queues.v1 import client
from zaqarclient.transport.errors import TransportError

from zaqar.bench.config import conf


def claim_delete(stats, test_duration, ttl, grace, limit):
    """Consumer Worker

    The Consumer Worker continuously claims and deletes messages
    for the specified duration. The time taken for each claim and
    delete is recorded for calculating throughput and latency.
    """

    cli = client.Client(conf.server_url)
    queue = cli.queue(conf.queue_prefix + '1')
    end = time.time() + test_duration
    claim_total_elapsed = 0
    delete_total_elapsed = 0
    total_failed_requests = 0
    claim_total_requests = 0
    delete_total_requests = 0

    while time.time() < end:
        try:
            marktime.start('claim_message')

            claim = queue.claim(ttl=ttl, grace=grace, limit=limit)

            claim_total_elapsed += marktime.stop('claim_message').seconds
            claim_total_requests += 1

        except TransportError as ex:
            sys.stderr.write("Could not claim messages : {0}\n".format(ex))
            total_failed_requests += 1

        else:
            for msg in claim:
                try:
                    marktime.start('delete_message')

                    msg.delete()

                    elapsed = marktime.stop('delete_message').seconds
                    delete_total_elapsed += elapsed
                    delete_total_requests += 1

                except TransportError as ex:
                    msg = "Could not delete messages: {0}\n".format(ex)
                    sys.stderr.write(msg)
                    total_failed_requests += 1

    total_requests = (claim_total_requests +
                      delete_total_requests +
                      total_failed_requests)

    stats.put({
        'total_requests': total_requests,
        'claim_total_requests': claim_total_requests,
        'delete_total_requests': delete_total_requests,
        'claim_total_elapsed': claim_total_elapsed,
        'delete_total_elapsed': delete_total_elapsed,
    })


{'fraction_non_alphanumeric': 0.07230795907750998, 'fraction_numerical': 0.007282816022195249, 'mean_word_length': 2.765013054830287, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4704585', 'n_tokens_mistral': 1593, 'n_tokens_neox': 1509, 'n_words': 494}