Upgrade to Pro — share decks privately, control downloads, hide ads and more …

Persistent Queues with Tape

Persistent Queues with Tape

The best apps don't make users wait. They dispatch tasks to the background. Offloading tasks to the background is a great way to simulate the feeling of responsiveness.

But processing background tasks can be tricky. What happens if the device runs out of battery? What if the network is down? Persisting to disk helps you handle such edge cases. Enter Tape, a collection of queue related classes.

This talk will primarily be about it’s core component, QueueFile — a lightning fast, transactional, persistent file-based FIFO queue. We'll dig into it's implementation and see how it guarantees both reliability and efficiency. And you'll learn to make the most of QueueFile with real world examples.

Prateek Srivastava

September 30, 2016
Tweet

More Decks by Prateek Srivastava

Other Decks in Programming

Transcript

  1. curl -X "POST" "https://api.segment.io/v1/track" \ -H "Content-Type: application/json" \ -u

    apiKey: \ -d "{\"event\":\"hello world\",\"userId\":\"prateek\"}"
  2. -RWD Open for reading and writing, as with "rw", and

    also require that every update to the file's content be written synchronously to the underlying storage device.
  3. @jake 16 16 1 4096 Last Element Position First Element

    Position Count File Size 8 bytes 8 bytes 4 bytes 63 bits
  4. @jake 16 16 1 4096 Last Element Position First Element

    Position Count File Size 8 bytes 8 bytes 4 bytes 63 bits
  5. 4096 1 32 32 26 Element 1 10 4096 1

    32 32 Element 1 10 Element 2 26 4096 1 32 32 Element 1 10
  6. 4096 1 46 46 Element 1 10 Element 2 26

    4096 1 46 46 Element 2 26
  7. // Read and remove the data at the // head

    of the queue. byte[] element = queueFile.remove(); // Do something with the data. process(element);
  8. // Read and remove the data at the // head

    of the queue. byte[] element = queueFile.remove(); // Do something with the data. process(element);
  9. // Read data at the head. byte[] element = queueFile.peek();

    // Do something with the data. process(element); // Remove data at the head. queueFile.remove();
  10. try { connection.close(); } catch (IOException e) { return; //

    Network error, retry later. } int responseCode = connection.getResponseCode(); if (responseCode >= 500) { return; // Server error, retry later. } // Client error, e.g. invalid JSON. Don’t retry. if (responseCode >= 400) { log("server rejected message"); } queueFile.remove();
  11. @master class QueueFile implements Iterable<byte[]> { void add(byte[] d, int

    offset, int count); boolean isEmpty(); void remove(int n); void clear(); }
  12. class QueueFile implements Iterable<byte[]> { void add(byte[] d, int offset,

    int count); boolean isEmpty(); void remove(int n); void clear(); } @master
  13. class QueueFile implements Iterable<byte[]> { void add(byte[] d, int offset,

    int count); boolean isEmpty(); void remove(int n); void clear(); } @master
  14. class QueueFile implements Iterable<byte[]> { void add(byte[] d, int offset,

    int count); boolean isEmpty(); void remove(int n); void clear(); } @master
  15. @master class QueueFile implements Iterable<byte[]> { void add(byte[] d, int

    offset, int count); boolean isEmpty(); void remove(int n); void clear(); }
  16. // Read the data at the head of the queue.

    List<byte[]> data = new ArrayList<>(); for (byte[] d : queueFile) { data.add(d); } // Do something with the elements. process(data); // Remove elements from the queue. queueFile.remove(data.size());
  17. HttpURLConnection connection = open(“/v1/import"); int count = 0; int size

    = 0; for (byte[] data : queueFile) { if (size + data.length > MAX_LIMIT) { break; } write(data, connection); count++; size += data.length; } connection.close(); queueFile.remove(count);
  18. @master abstract class ObjectQueue<T> implements Iterable<T> { abstract void add(T

    entry); abstract T peek(); List<T> peek(int max); List<T> asList(); void remove(); abstract void remove(int n); void clear(); abstract int size(); }
  19. @master abstract class ObjectQueue<T> implements Iterable<T> { abstract void add(T

    entry); abstract T peek(); List<T> peek(int max); List<T> asList(); void remove(); abstract void remove(int n); void clear(); abstract int size(); }
  20. @master abstract class ObjectQueue<T> implements Iterable<T> { abstract void add(T

    entry); abstract T peek(); List<T> peek(int max); List<T> asList(); void remove(); abstract void remove(int n); void clear(); abstract int size(); }
  21. // Read data at the head. Pojo pojo = queue.peek();

    // Do something with the data. process(pojo); // Remove data at the head. queue.remove();
  22. // Read the data at the head of the queue.

    List<Pojo> data = new ArrayList<>(); for (Pojo pojo : queue) { data.add(pojo); } // Do something with the elements. process(data); // Remove elements from the queue. queue.remove(data.size());