1
\$\begingroup\$

I have an Android app that collects data in n number of SQLite databases. The user decides the value of n in runtime. A Service is in charge of sending that data to our server. We used to do this with an IntentService but since there were some "HighFrequency" databases that were 100x bigger than the smaller ones, we decided to use MultipleThreads to let the small ones sync at the same time as the big ones. The sync session on a single thread could take up to 7 hours.

We have 3 HandlerThread objects that receive a message containing the db to be synced. Within the handler a ExecutorService object is used to do the sync per db on a different thread and wait for it.

One HanlderThread is used for small databases and the other 2 HandlerThread are used to split the HighFrequency databases (the bigger ones).

We came up with this implementation but I am not sure if it is the best one or if we are at risk of memory leaks. Thanks for your opinion!

This the original class and this is a reduced version:

public class WebserviceHelper extends Service {

    public static final String EXTRA_TABLE = "table";
    public static final String EXTRA_FIELDS = "fields";
    public static final String EXTRA_CONTENT_URI = "uri";

    private static long sync_start = 0;
    private int notificationID = 0;
    private static int total_rows_synced = 0;

    private SyncQueue mSyncFastQueue;
    private Looper mServiceLooperFastQueue;
    ExecutorService executorFastQueue;

    private SyncQueue mSyncSlowQueueA;
    private Looper mServiceLooperSlowQueueA;
    ExecutorService executorSlowQueueA;

    private SyncQueue mSyncSlowQueueB;
    private Looper mServiceLooperSlowQueueB;
    ExecutorService executorSlowQueueB;

    private boolean nextSlowQueue = false;

    private Map<String, Boolean> SYNCED_TABLES;
    private static final ArrayList<String> highFrequencySensors = new ArrayList<>();

    // Handler that receives messages from the thread
    private final class SyncQueue extends Handler {
        ExecutorService executor;

        public SyncQueue(Looper looper, ExecutorService executor) {
            super(looper);
            // One thread to sync data with the server
            this.executor = executor;
        }

        @Override
        public void handleMessage(Message msg) {
            Bundle bundle = msg.getData();
            Context context = (Context) msg.obj;
            SyncTable syncTable = new SyncTable(context, bundle.getBoolean("DEBUG"), bundle.getString("DATABASE_TABLE"), bundle.getString("TABLES_FIELDS"), bundle.getString("ACTION"), bundle.getString("CONTENT_URI_STRING"), bundle.getString("DEVICE_ID"), bundle.getString("WEBSERVER"), bundle.getBoolean("WEBSERVICE_SIMPLE"), bundle.getBoolean("WEBSERVICE_REMOVE_DATA"), bundle.getInt("MAX_POST_SIZE"), bundle.getInt("notificationID"));
            try {
                if (!executor.isTerminated())
                    executor.submit(syncTable).get();
            } catch (InterruptedException | ExecutionException e) {
                if(Aware.DEBUG)
                    Log.e(Aware.TAG, e.getMessage());
            }

            // Stop the service using the startId, so that we don't stop
            // the service in the middle of handling another job
            stopSelf(msg.arg1);
        }
    }

    @Override
    public void onCreate() {
        super.onCreate();

        highFrequencySensors.add("accelerometer");
        highFrequencySensors.add("gyroscope");
        highFrequencySensors.add("barometer");
        highFrequencySensors.add("gravity");
        highFrequencySensors.add("linear_accelerometer");
        highFrequencySensors.add("magnetometer");
        highFrequencySensors.add("rotation");
        highFrequencySensors.add("temperature");
        highFrequencySensors.add("proximity");

        notificationID = 0;
        SYNCED_TABLES = Collections.synchronizedMap(new HashMap<String, Boolean>());

        HandlerThread threadFast = new HandlerThread("SyncFastQueue",
                android.os.Process.THREAD_PRIORITY_BACKGROUND);
        threadFast.start();

        HandlerThread threadSlowA = new HandlerThread("SyncSlowAQueue",
                android.os.Process.THREAD_PRIORITY_BACKGROUND);
        threadSlowA.start();

        HandlerThread threadSlowB = new HandlerThread("SyncSlowBQueue",
                android.os.Process.THREAD_PRIORITY_BACKGROUND);
        threadSlowB.start();

        // Get the HandlerThread's Looper and use it for our Handler
        mServiceLooperFastQueue = threadFast.getLooper();
        executorFastQueue = Executors.newSingleThreadExecutor();
        mSyncFastQueue = new SyncQueue(mServiceLooperFastQueue, executorFastQueue);

        mServiceLooperSlowQueueA = threadSlowA.getLooper();
        executorSlowQueueA = Executors.newSingleThreadExecutor();
        mSyncSlowQueueA = new SyncQueue(mServiceLooperSlowQueueA, executorSlowQueueA);

        mServiceLooperSlowQueueB = threadSlowB.getLooper();
        executorSlowQueueB = Executors.newSingleThreadExecutor();
        mSyncSlowQueueB = new SyncQueue(mServiceLooperSlowQueueB, executorSlowQueueB);

    }

    @Override
    public int onStartCommand(Intent intent, int flags, int startId) {

        String WEBSERVER = //
        int MAX_POST_SIZE = getBatchSize();

        if (intent.getAction().equals(ACTION_AWARE_WEBSERVICE_SYNC_TABLE)) {

            String table = intent.getStringExtra(EXTRA_TABLE);

            if (!SYNCED_TABLES.containsKey(table)) {
                SYNCED_TABLES.put(table, true);
                if(!highFrequencySensors.contains(table)){ //Non High Frequency sensors go together
                    Message msg = buildMessage(mSyncFastQueue, intent, DEBUG, DEVICE_ID, WEBSERVER, WEBSERVICE_SIMPLE, WEBSERVICE_REMOVE_DATA, MAX_POST_SIZE, startId, notificationID++);
                    mSyncFastQueue.sendMessage(msg);
                }
                else if(nextSlowQueue){ // High frequency sensors are split in two threads
                    Message msg = buildMessage(mSyncSlowQueueA, intent, DEBUG, DEVICE_ID, WEBSERVER, WEBSERVICE_SIMPLE, WEBSERVICE_REMOVE_DATA, MAX_POST_SIZE, startId, notificationID++);
                    mSyncSlowQueueA.sendMessage(msg);
                    nextSlowQueue = !nextSlowQueue;
                }
                else{
                    Message msg = buildMessage(mSyncSlowQueueB, intent, DEBUG, DEVICE_ID, WEBSERVER, WEBSERVICE_SIMPLE, WEBSERVICE_REMOVE_DATA, MAX_POST_SIZE, startId, notificationID++);
                    mSyncSlowQueueB.sendMessage(msg);
                    nextSlowQueue = !nextSlowQueue;
                }
            } else {
                if(Aware.DEBUG)
                    Log.d(Aware.TAG, "Tried to sync again for the same table: " + table);
            }

        }

        // If we get killed, after returning from here, restart
        return START_REDELIVER_INTENT;
    }

    private Message buildMessage(SyncQueue queue, Intent intent, boolean DEBUG, String DEVICE_ID, String WEBSERVER, boolean WEBSERVICE_SIMPLE, boolean WEBSERVICE_REMOVE_DATA, int MAX_POST_SIZE, int startId, int notificationID){
        Message msg = queue.obtainMessage();
        Bundle bundle = new Bundle();
        bundle.putBoolean("DEBUG", DEBUG);
        bundle.putString("DEVICE_ID", DEVICE_ID);
        bundle.putString("WEBSERVER", WEBSERVER);
        bundle.putBoolean("WEBSERVICE_SIMPLE", WEBSERVICE_SIMPLE);
        bundle.putBoolean("WEBSERVICE_REMOVE_DATA", WEBSERVICE_REMOVE_DATA);
        bundle.putInt("MAX_POST_SIZE", MAX_POST_SIZE);
        bundle.putInt("notificationID", notificationID);
        bundle.putString("DATABASE_TABLE", intent.getStringExtra(EXTRA_TABLE));
        bundle.putString("TABLES_FIELDS", intent.getStringExtra(EXTRA_FIELDS));
        bundle.putString("ACTION", intent.getAction());
        bundle.putString("CONTENT_URI_STRING", intent.getStringExtra(EXTRA_CONTENT_URI));
        msg.obj = getApplicationContext();
        msg.setData(bundle);
        msg.arg1 = startId;

        return msg;

    }

    /**
     * Asynchronously process the sync of all tables
     */
    private static class SyncTable implements Callable<String> {
        private Context mContext;
        private boolean DEBUG;
        private String TABLES_FIELDS;
        private String ACTION;
        private String CONTENT_URI_STRING;
        private String DEVICE_ID;
        private String DATABASE_TABLE;
        private String WEBSERVER;
        private String protocol;
        private boolean WEBSERVICE_SIMPLE;
        private boolean WEBSERVICE_REMOVE_DATA;
        private int MAX_POST_SIZE;
        private int NOTIFICATION_ID;

        SyncTable(Context c, boolean debug, String table, String fields, String action, String uri, String deviceID, String webServer, boolean webServiceSimple, boolean webServiceRemoveData, int maxPostSize, int notificationID) {
            //Initialise variables
        }


        @Override
        public String call() throws Exception {

            //Send database records to the server using batches of 5k and a HTTPS connection

            return Thread.currentThread().getName();
        }
    }

    @Override
    public IBinder onBind(Intent intent) {
        // We don't provide binding, so return null
        return null;
    }

    @Override
    public void onDestroy() {

        mServiceLooperFastQueue.quitSafely();
        mServiceLooperSlowQueueA.quitSafely();
        mServiceLooperSlowQueueB.quitSafely();
        executorFastQueue.shutdown();
        executorSlowQueueA.shutdown();
        executorSlowQueueB.shutdown();
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.