Revert "Celery experiment (#720)" (#726)

This reverts commit 382e92c.

.travis.yml

@@ -12,7 +12,6 @@ node_js:
 
 services:
   - postgresql
-  - rabbitmq
 
 sudo: false
 
@@ -24,14 +23,10 @@ before_script:
   - psql -c 'create database anyway;' -U postgres
   - export DATABASE_URL='postgresql://postgres@localhost/anyway'
   - python models.py
-  - celery worker -A clusters_calculator -D
 
 script:
   - pylint -j $(nproc) *.py tests
   - eslint static/js --ignore-path=static/js/.eslintignore
   - python process.py
   - python united.py --light
   - pytest tests
-
-after_script:
-  - kill $(cat celeryd.pid)

clusters_calculator.py

@@ -1,25 +1,27 @@
-import itertools
-from celery import Celery, group
 from models import Marker
 from static.pymapcluster import calculate_clusters
+import logging
+import concurrent.futures
 import multiprocessing
 
 
-celery_app = Celery('tasks', backend='rpc://', broker='pyamqp://guest@localhost//')
+def retrieve_clusters(**kwargs):
+    marker_boxes = divide_to_boxes(kwargs['ne_lat'], kwargs['ne_lng'], kwargs['sw_lat'], kwargs['sw_lng'])
+    result_futures = []
+    logging.info('number of cores: ' + str(multiprocessing.cpu_count()))
+    with concurrent.futures.ThreadPoolExecutor(max_workers=multiprocessing.cpu_count()) as executor:
+        for marker_box in marker_boxes:
 
-@celery_app.task
-def calculate_marker_box(kwargs, marker_box):
-    kwargs.update(marker_box)
-    markers_in_box = Marker.bounding_box_query(**kwargs).markers.all()
-    return calculate_clusters(markers_in_box, kwargs['zoom'])
+            kwargs.update(marker_box)
+            markers_in_box = Marker.bounding_box_query(**kwargs).markers.all()
+            result_futures.append(executor.submit(calculate_clusters, markers_in_box, kwargs['zoom']))
 
+    completed_futures = concurrent.futures.wait(result_futures)
+    result = []
+    for future in completed_futures.done:
+        result.extend(future.result())
 
-def retrieve_clusters(**kwargs):
-    marker_boxes = divide_to_boxes(kwargs['ne_lat'], kwargs['ne_lng'], kwargs['sw_lat'], kwargs['sw_lng'])
-    job = group([calculate_marker_box.s(kwargs, marker_box) for marker_box in marker_boxes])
-    result = job.apply_async()
-    result.join()
-    return list(itertools.chain.from_iterable(result.get()))
+    return result
 
 
 def divide_to_boxes(ne_lat, ne_lng, sw_lat, sw_lng):

requirements.txt

@@ -20,10 +20,10 @@ Flask-Admin==1.1.0
 Flask-Security==1.7.5
 WTForms==2.0.2
 sendgrid==1.4.0
+futures==2.1.6
 python-dateutil==2.4.2
 alembic==0.8.2
 apscheduler==2.1.2
 Flask-Compress==1.3.0
 rauth==0.7.2
-requests==2.4.3
-celery
+requests==2.4.3