Skip to content

added --batchsize and --interval #40

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Mar 31, 2017
Merged

added --batchsize and --interval #40

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
857866a
added --batchsize and --interval
masayang Mar 24, 2017
75db118
fixed failed tests
masayang Mar 24, 2017
9b3a69a
why pep8 rejects this code?
masayang Mar 28, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,3 +6,6 @@ dist/*
*.egg-info/*
build/*
htmlcov/*
*~
*#

26 changes: 23 additions & 3 deletions pyqs/main.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -72,20 +72,40 @@ def main():
action="store",
)

parser.add_argument(
"--interval",
dest="interval",
type=float,
default=0.01,
help='interval',
action="store",
)

parser.add_argument(
"--batchsize",
dest="batchsize",
type=int,
default=10,
help='batchsize',
action="store",
)

args = parser.parse_args()

_main(queue_prefixes=args.queues,
concurrency=args.concurrency,
logging_level=args.logging_level,
region=args.region,
access_key_id=args.access_key_id,
secret_access_key=args.secret_access_key
secret_access_key=args.secret_access_key,
interval=args.interval,
batchsize=args.batchsize
)


def _main(queue_prefixes, concurrency=5, logging_level="WARN", region='us-east-1', access_key_id=None, secret_access_key=None):
def _main(queue_prefixes, concurrency=5, logging_level="WARN", region='us-east-1', access_key_id=None, secret_access_key=None, interval=1, batchsize=10):
logging.basicConfig(format="[%(levelname)s]: %(message)s", level=getattr(logging, logging_level))
logger.info("Starting PyQS version {}".format(__version__))
manager = ManagerWorker(queue_prefixes, concurrency, region=region, access_key_id=access_key_id, secret_access_key=secret_access_key)
manager = ManagerWorker(queue_prefixes, concurrency, interval, batchsize, region=region, access_key_id=access_key_id, secret_access_key=secret_access_key)
manager.start()
manager.sleep()
27 changes: 19 additions & 8 deletions pyqs/worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,11 +52,16 @@ def parent_is_alive(self):

class ReadWorker(BaseWorker):

def __init__(self, sqs_queue, internal_queue, *args, **kwargs):
def __init__(self, sqs_queue, internal_queue, batchsize, *args, **kwargs):
super(ReadWorker, self).__init__(*args, **kwargs)
self.sqs_queue = sqs_queue
self.visibility_timeout = self.sqs_queue.get_timeout()
self.internal_queue = internal_queue
self.batchsize = batchsize
if batchsize > MESSAGE_DOWNLOAD_BATCH_SIZE:
self.batchsize = MESSAGE_DOWNLOAD_BATCH_SIZE
if batchsize <= 0:
self.batchsize = 1

def run(self):
# Set the child process to not receive any keyboard interrupts
Expand All @@ -67,7 +72,8 @@ def run(self):
self.read_message()

def read_message(self):
messages = self.sqs_queue.get_messages(MESSAGE_DOWNLOAD_BATCH_SIZE, wait_time_seconds=LONG_POLLING_INTERVAL)

messages = self.sqs_queue.get_messages(self.batchsize, wait_time_seconds=LONG_POLLING_INTERVAL)
logger.info("Successfully got {} messages from SQS queue {}".format(len(messages), self.sqs_queue.name)) # noqa
start = time.time()
for message in messages:
Expand Down Expand Up @@ -98,14 +104,16 @@ def read_message(self):

class ProcessWorker(BaseWorker):

def __init__(self, internal_queue, connection_args=None, *args, **kwargs):
def __init__(self, internal_queue, interval, connection_args=None, *args, **kwargs):
super(ProcessWorker, self).__init__(*args, **kwargs)
if connection_args is None:
self.conn = get_conn()
else:
self.conn = get_conn(**connection_args)
self.internal_queue = internal_queue
self.interval = interval
self._messages_to_process_before_shutdown = 100


def run(self):
# Set the child process to not receive any keyboard interrupts
Expand All @@ -117,6 +125,7 @@ def run(self):
processed = self.process_message()
if processed:
messages_processed += 1
time.sleep(self.interval)
else:
# If we have no messages wait a moment before rechecking.
time.sleep(0.001)
Expand Down Expand Up @@ -187,12 +196,14 @@ def process_message(self):

class ManagerWorker(object):

def __init__(self, queue_prefixes, worker_concurrency, region='us-east-1', access_key_id=None, secret_access_key=None):
def __init__(self, queue_prefixes, worker_concurrency, interval, batchsize, region='us-east-1', access_key_id=None, secret_access_key=None):
self.connection_args = {
"region": region,
"access_key_id": access_key_id,
"secret_access_key": secret_access_key,
}
self.interval = interval
self.batchsize = batchsize
self.load_queue_prefixes(queue_prefixes)
self.queues = self.get_queues_from_queue_prefixes(self.queue_prefixes)
self.setup_internal_queue(worker_concurrency)
Expand All @@ -209,11 +220,11 @@ def _register_signals(self):

def _initialize_reader_children(self):
for queue in self.queues:
self.reader_children.append(ReadWorker(queue, self.internal_queue))
self.reader_children.append(ReadWorker(queue, self.internal_queue, self.batchsize))

def _initialize_worker_children(self, number):
for index in range(number):
self.worker_children.append(ProcessWorker(self.internal_queue, connection_args=self.connection_args))
self.worker_children.append(ProcessWorker(self.internal_queue, self.interval, connection_args=self.connection_args))

def load_queue_prefixes(self, queue_prefixes):
self.queue_prefixes = queue_prefixes
Expand Down Expand Up @@ -291,7 +302,7 @@ def _replace_reader_children(self):
logger.info("Reader Process {} is no longer responding, spawning a new reader.".format(reader.pid))
queue = reader.sqs_queue
self.reader_children.pop(index)
worker = ReadWorker(queue, self.internal_queue)
worker = ReadWorker(queue, self.internal_queue, self.batchsize)
worker.start()
self.reader_children.append(worker)

Expand All @@ -300,6 +311,6 @@ def _replace_worker_children(self):
if not worker.is_alive():
logger.info("Worker Process {} is no longer responding, spawning a new worker.".format(worker.pid))
self.worker_children.pop(index)
worker = ProcessWorker(self.internal_queue, connection_args=self.connection_args)
worker = ProcessWorker(self.internal_queue, self.interval, connection_args=self.connection_args)
worker.start()
self.worker_children.append(worker)
22 changes: 13 additions & 9 deletions tests/test_manager_worker.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ def test_manager_worker_create_proper_children_workers():
conn = boto.connect_sqs()
conn.create_queue("email")

manager = ManagerWorker(queue_prefixes=['email'], worker_concurrency=3)
manager = ManagerWorker(queue_prefixes=['email'], worker_concurrency=3, interval=2, batchsize=10)

len(manager.reader_children).should.equal(1)
len(manager.worker_children).should.equal(3)
Expand All @@ -35,7 +35,7 @@ def test_manager_worker_with_queue_prefix():
conn.create_queue("email.foobar")
conn.create_queue("email.baz")

manager = ManagerWorker(queue_prefixes=['email.*'], worker_concurrency=1)
manager = ManagerWorker(queue_prefixes=['email.*'], worker_concurrency=1, interval=1, batchsize=10)

len(manager.reader_children).should.equal(2)
children = manager.reader_children
Expand All @@ -54,7 +54,7 @@ def test_manager_start_and_stop():
conn = boto.connect_sqs()
conn.create_queue("email")

manager = ManagerWorker(queue_prefixes=['email'], worker_concurrency=2)
manager = ManagerWorker(queue_prefixes=['email'], worker_concurrency=2, interval=1, batchsize=10)

len(manager.worker_children).should.equal(2)

Expand All @@ -80,7 +80,7 @@ def test_main_method(ManagerWorker):
"""
_main(["email1", "email2"], concurrency=2)

ManagerWorker.assert_called_once_with(['email1', 'email2'], 2, region='us-east-1', secret_access_key=None, access_key_id=None)
ManagerWorker.assert_called_once_with(['email1', 'email2'], 2, 1, 10, region='us-east-1', secret_access_key=None, access_key_id=None)
ManagerWorker.return_value.start.assert_called_once_with()


Expand All @@ -93,6 +93,8 @@ def test_real_main_method(ArgumentParser, _main):
"""
ArgumentParser.return_value.parse_args.return_value = Mock(concurrency=3,
queues=["email1"],
interval=1,
batchsize=10,
logging_level="WARN",
region='us-east-1',
access_key_id=None,
Expand All @@ -101,6 +103,8 @@ def test_real_main_method(ArgumentParser, _main):

_main.assert_called_once_with(queue_prefixes=['email1'],
concurrency=3,
interval=1,
batchsize=10,
logging_level="WARN",
region='us-east-1',
access_key_id=None,
Expand All @@ -118,7 +122,7 @@ def test_master_spawns_worker_processes():
conn.create_queue("tester")

# Setup Manager
manager = ManagerWorker(["tester"], 1)
manager = ManagerWorker(["tester"], 1, 1, 10)
manager.start()

# Check Workers
Expand All @@ -143,7 +147,7 @@ def test_master_replaces_reader_processes():
conn.create_queue("tester")

# Setup Manager
manager = ManagerWorker(queue_prefixes=["tester"], worker_concurrency=1)
manager = ManagerWorker(queue_prefixes=["tester"], worker_concurrency=1, interval=1, batchsize=10)
manager.start()

# Get Reader PID
Expand Down Expand Up @@ -177,7 +181,7 @@ def test_master_counts_processes():
conn.create_queue("tester")

# Setup Manager
manager = ManagerWorker(["tester"], 2)
manager = ManagerWorker(["tester"], 2, 1, 10)
manager.start()

# Check Workers
Expand All @@ -203,7 +207,7 @@ def test_master_replaces_worker_processes():
conn.create_queue("tester")

# Setup Manager
manager = ManagerWorker(queue_prefixes=["tester"], worker_concurrency=1)
manager = ManagerWorker(queue_prefixes=["tester"], worker_concurrency=1, interval=1, batchsize=10)
manager.start()

# Get Worker PID
Expand Down Expand Up @@ -240,7 +244,7 @@ def process_counts():
os.kill(os.getpid(), signal.SIGTERM)

# Setup Manager
manager = ManagerWorker(queue_prefixes=["tester"], worker_concurrency=1)
manager = ManagerWorker(queue_prefixes=["tester"], worker_concurrency=1, interval=1, batchsize=10)
manager.process_counts = process_counts
manager._graceful_shutdown = MagicMock()

Expand Down
Loading