Data concurrency in Android Service with sensor data

Question

My application is running a Service that holds a BLE connection to a multi-sensor wristband. The Serviceimplements some callback methods for the wristband SDK which are called several times every seconds with new data.

I want to put these data, from the different sensors, within the same Observation object relative to its timestamp. All Observation objects are pushed to a backend server every 60 seconds, sensor data is put together to reduce the overhead in sending these Observation objects.

What I'm doing now is presented in the code snippet below. My problem is that the while-loop in observationFetcher completely blocks the application. Is there any other approaches for synchronizing these sensor data without using a block while-loop?

    observationFetcher = new Runnable() {
        @Override
        public void run() {
            while (isRecording) {
                if (lastMillis != currentMillis) {
                    Observation obs = sm.getValues();
                    obs.setPropertyAsString("gateway.id", UUID);
                    observations.add(obs);
                    lastMillis = currentMillis;
                }
            }
        }
    };

public void didReceiveGSR(float gsr, double timestamp) {
    long t = System.currentTimeMillis() / 1000;

    sm.setGsrValue(t, gsr);
    currentMillis = t;
}

public void didReceiveIBI(float ibi, double timestamp) {
    sm.setIbiValue(ibi);
}

sm is an object with synchronized methods for putting all the sensor data within the same second together.

Answer 1

Please correct me if I'm wrong, but I don't see a reason to waste CPU time infinity iterating. Of course, I don't see the entire code and your API may not allow you to do something, but I would implement the data processing in following way:

final class Observation {
    private float gsr;
    private float ibi;

    public Observation(float gsr, float ibi) {
        this.gsr = gsr;
        this.ibi = ibi;
    }

    // getters & setters

}

public final class Observations {
    private final ConcurrentHashMap<Long, Observation> observations = new ConcurrentHashMap<>();

    public void insertGsrValue(long timestamp, float gsr) {
        for (;;) {
            Observation observation = observations.get(timestamp);
            if (observation == null) {
                observation = observations.putIfAbsent(timestamp, new Observation(gsr, 0.0f));
                if (observation == null) {
                    return;
                }
            }
            if (observations.replace(timestamp, observation, new Observation(gsr, observation.getIbi()))) {
                return;
            }
        }
    }

    public void insertIbiValue(long timestamp, float ibi) {
        for (;;) {
            Observation observation = observations.get(timestamp);
            if (observation == null) {
                observation = observations.putIfAbsent(timestamp, new Observation(0.0f, ibi));
                if (observation == null) {
                    return;
                }
            }
            if (observations.replace(timestamp, observation, new Observation(observation.getGsr(), ibi))) {
                return;
            }
        }
    }

    public List<Observation> getObservations() {
        return new ArrayList<>(observations.values());
    }

    public void clear() {
        observations.clear();
    }

}

public final class ObservationService extends Service {
    private final Observations observations = new Observations();
    private volatile long currentMillis;
    private HandlerThread handlerThread;
    private Handler handler;

    @Override
    public void onCreate() {
        super.onCreate();
        handlerThread = new HandlerThread("observations_sender_thread");
        handlerThread.start();
        handler = new Handler(handlerThread.getLooper());
        handler.postDelayed(new Runnable() {
            @Override
            public void run() {
                sendData();
                handler.postDelayed(this, TimeUnit.SECONDS.toMillis(60));
            }
        }, TimeUnit.SECONDS.toMillis(60));
    }

    @Override
    public void onDestroy() {
        handlerThread.quit();
    }

    private void sendData() {
        List<Observation> observationList = observations.getObservations();
        observations.clear();
        // send observation list somehow
    }

    public void didReceiveGSR(float gsr, double timestamp) {
        // assuming this is called on a worker thread
        long t = System.currentTimeMillis() / 1000;
        observations.insertGsrValue(t, gsr);
        currentMillis = t;
    }

    public void didReceiveIBI(float ibi, double timestamp) {
        // assuming this is called on a worker thread
        observations.insertIbiValue(currentMillis, ibi);
    }

    @Nullable
    @Override
    public IBinder onBind(Intent intent) {
        return null;
    }
}

So what this code does is insert new values from sensors into a hash map and send it somewhere every 60 seconds. This code is still not perfect as there is a problem with concurrency. For example, if 2 gsr values come first and then one ibi value, then we will lose the first gsr value.

Anyway, this code should give an idea how you can avoid blocking the thread and store the data concurrency.

Please do let me know if you have any questions regarding the code.