Таразанов Максим, ИТМО, М4139

.gitignore

@@ -5,6 +5,8 @@
 build
 
 .idea
+.obsidian
+
 
 # Compiled source #
 ###################

src/main/java/ru/vk/itmo/test/reference/ReferenceServer.java

@@ -1,6 +1,12 @@
 package ru.vk.itmo.test.reference;
 
-import one.nio.http.*;
+import one.nio.http.HttpServer;
+import one.nio.http.HttpServerConfig;
+import one.nio.http.HttpSession;
+import one.nio.http.Param;
+import one.nio.http.Path;
+import one.nio.http.Request;
+import one.nio.http.Response;
 import one.nio.server.AcceptorConfig;
 import one.nio.util.Utf8;
 import org.slf4j.Logger;
@@ -9,7 +15,6 @@
 import ru.vk.itmo.dao.BaseEntry;
 import ru.vk.itmo.dao.Dao;
 import ru.vk.itmo.dao.Entry;
-
 import java.io.IOException;
 import java.lang.foreign.MemorySegment;
 import java.lang.foreign.ValueLayout;

src/main/java/ru/vk/itmo/test/reference/Server.java

@@ -5,7 +5,6 @@
 import ru.vk.itmo.test.reference.dao.ReferenceDao;
 
 import java.io.IOException;
-import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.List;
 

src/main/java/ru/vk/itmo/test/tarazanovmaxim/MyFactory.java

@@ -0,0 +1,14 @@
+package ru.vk.itmo.test.tarazanovmaxim;
+
+import ru.vk.itmo.Service;
+import ru.vk.itmo.ServiceConfig;
+import ru.vk.itmo.test.ServiceFactory;
+
+@ServiceFactory(stage = 1)
+public class MyFactory implements ServiceFactory.Factory {
+
+    @Override
+    public Service create(ServiceConfig config) {
+        return new MyService(config);
+    }
+}

src/main/java/ru/vk/itmo/test/tarazanovmaxim/MyServer.java

@@ -0,0 +1,178 @@
+package ru.vk.itmo.test.tarazanovmaxim;
+
+import one.nio.http.HttpServer;
+import one.nio.http.HttpServerConfig;
+import one.nio.http.HttpSession;
+import one.nio.http.Param;
+import one.nio.http.Path;
+import one.nio.http.Request;
+import one.nio.http.RequestMethod;
+import one.nio.http.Response;
+import one.nio.server.AcceptorConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import ru.vk.itmo.ServiceConfig;
+import ru.vk.itmo.dao.BaseEntry;
+import ru.vk.itmo.dao.Config;
+import ru.vk.itmo.dao.Entry;
+import ru.vk.itmo.test.tarazanovmaxim.dao.ReferenceDao;
+
+import java.io.IOException;
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.TimeUnit;
+
+public class MyServer extends HttpServer {
+
+    private final ReferenceDao dao;
+    private static final long FLUSH_THRESHOLD_BYTES = 1 << 20;
+    private static final String PATH = "/v0/entity";
+    private static final long REQUEST_TTL = TimeUnit.SECONDS.toNanos(100);
+    private final ExecutorService executorService;
+    private static final Logger logger = LoggerFactory.getLogger(MyServer.class);
+
+    public MyServer(ServiceConfig config) throws IOException {
+        super(createServerConfig(config));
+        dao = new ReferenceDao(new Config(config.workingDir(), FLUSH_THRESHOLD_BYTES));
+        executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() / 2 + 1);
+    }
+
+    private static HttpServerConfig createServerConfig(ServiceConfig serviceConfig) {
+        HttpServerConfig httpServerConfig = new HttpServerConfig();
+        AcceptorConfig acceptorConfig = new AcceptorConfig();
+
+        acceptorConfig.port = serviceConfig.selfPort();
+        acceptorConfig.reusePort = true;
+
+        httpServerConfig.acceptors = new AcceptorConfig[]{acceptorConfig};
+        httpServerConfig.closeSessions = true;
+
+        return httpServerConfig;
+    }
+
+    private static MemorySegment toMemorySegment(String string) {
+        return MemorySegment.ofArray(string.getBytes(StandardCharsets.UTF_8));
+    }
+
+    public void close() {
+        executorService.close();
+        try {
+            dao.close();
+        } catch (IOException e) {
+            logger.error("IOException in close()->dao.close()");
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Path(PATH)
+    @RequestMethod(Request.METHOD_GET)
+    public final Response get(@Param(value = "id", required = true) String id) {
+        MemorySegment key =
+                (id == null || id.isEmpty())
+                ? null
+                : toMemorySegment(id);
+
+        if (key == null) {
+            return new Response(Response.BAD_REQUEST, Response.EMPTY);
+        }
+
+        Entry<MemorySegment> entry = dao.get(key);
+
+        if (entry == null) {
+            return new Response(Response.NOT_FOUND, Response.EMPTY);
+        }
+
+        return Response.ok(entry.value().toArray(ValueLayout.JAVA_BYTE));
+    }
+
+    @Path(PATH)
+    @RequestMethod(Request.METHOD_PUT)
+    public final Response put(@Param(value = "id", required = true) String id, Request request) {
+        MemorySegment key =
+                (id == null || id.isEmpty())
+                ? null
+                : toMemorySegment(id);
+
+        if (key == null) {
+            return new Response(Response.BAD_REQUEST, Response.EMPTY);
+        }
+
+        Entry<MemorySegment> entry = new BaseEntry<>(
+                key,
+                MemorySegment.ofArray(request.getBody())
+        );
+
+        dao.upsert(entry);
+
+        return new Response(Response.CREATED, Response.EMPTY);
+    }
+
+    @Path(PATH)
+    @RequestMethod(Request.METHOD_DELETE)
+    public final Response delete(@Param(value = "id", required = true) String id) {
+        MemorySegment key =
+                (id == null || id.isEmpty())
+                ? null
+                : toMemorySegment(id);
+
+        if (key == null) {
+            return new Response(Response.BAD_REQUEST, Response.EMPTY);
+        }
+
+        dao.upsert(new BaseEntry<>(key, null));
+
+        return new Response(Response.ACCEPTED, Response.EMPTY);
+    }
+
+    @Path(PATH)
+    public Response otherMethod() {
+        return new Response(Response.METHOD_NOT_ALLOWED, Response.EMPTY);
+    }
+
+    @Override
+    public void handleDefault(Request request, HttpSession session) {
+        Response response = new Response(Response.BAD_REQUEST, Response.EMPTY);
+        sendResponse(response, session);
+    }
+
+    @Override
+    public void handleRequest(Request request, HttpSession session) {
+        try {
+            long startTime = System.nanoTime();
+            executorService.execute(() -> {
+                if (System.nanoTime() > startTime + REQUEST_TTL) {
+                    sendResponse(new Response(Response.REQUEST_TIMEOUT, Response.EMPTY), session);
+                    return;
+                }
+                try {
+                    super.handleRequest(request, session);
+                } catch (Exception e) {
+                    logger.error("IOException in handleRequest->executorService.execute()");
+                    System.out.println(e.getClass());
+                    sendResponse(
+                        new Response(
+                            e.getClass() == IOException.class ? Response.INTERNAL_ERROR : Response.BAD_REQUEST,
+                            Response.EMPTY
+                        ),
+                        session
+                    );
+                }
+            });
+        } catch (RejectedExecutionException e) {
+            logger.error("RejectedExecutionException in handleRequest: " + request + session);
+            sendResponse(new Response("429 Too Many Requests", Response.EMPTY), session);
+        }
+    }
+
+    public void sendResponse(Response response, HttpSession session) {
+        try {
+            session.sendResponse(response);
+        } catch (IOException e) {
+            logger.error("IOException in sendResponse: " + response + session);
+        }
+    }
+}

src/main/java/ru/vk/itmo/test/tarazanovmaxim/MyService.java

@@ -0,0 +1,31 @@
+package ru.vk.itmo.test.tarazanovmaxim;
+
+import ru.vk.itmo.Service;
+import ru.vk.itmo.ServiceConfig;
+
+import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
+
+public class MyService implements Service {
+    private MyServer server;
+
+    private final ServiceConfig config;
+
+    public MyService(ServiceConfig config) {
+        this.config = config;
+    }
+
+    @Override
+    public CompletableFuture<Void> start() throws IOException {
+        server = new MyServer(config);
+        server.start();
+        return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public CompletableFuture<Void> stop() throws IOException {
+        server.stop();
+        server.close();
+        return CompletableFuture.completedFuture(null);
+    }
+}

src/main/java/ru/vk/itmo/test/tarazanovmaxim/ServerMain.java

@@ -0,0 +1,27 @@
+package ru.vk.itmo.test.tarazanovmaxim;
+
+import ru.vk.itmo.ServiceConfig;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.List;
+
+public class ServerMain {
+
+    private ServerMain() {
+
+    }
+
+    public static void main(String[] args) throws IOException {
+        MyServer server = new MyServer(
+            new ServiceConfig(
+                8080,
+                "http://localhost",
+                List.of("http://localhost"),
+                Files.createTempDirectory("dao")
+            )
+        );
+
+        server.start();
+    }
+}

src/main/java/ru/vk/itmo/test/tarazanovmaxim/dao/ByteArraySegment.java

@@ -0,0 +1,48 @@
+package ru.vk.itmo.test.tarazanovmaxim.dao;
+
+import java.io.IOException;
+import java.lang.foreign.MemorySegment;
+import java.nio.ByteBuffer;
+
+/**
+ * Growable buffer with {@link ByteBuffer} and {@link MemorySegment} interface.
+ *
+ * @author incubos
+ */
+final class ByteArraySegment {
+    private byte[] array;
+    private MemorySegment segment;
+
+    ByteArraySegment(final int capacity) {
+        this.array = new byte[capacity];
+        this.segment = MemorySegment.ofArray(array);
+    }
+
+    void withArray(final ArrayConsumer consumer) throws IOException {
+        consumer.process(array);
+    }
+
+    MemorySegment segment() {
+        return segment;
+    }
+
+    void ensureCapacity(final long size) {
+        if (size > Integer.MAX_VALUE) {
+            throw new IllegalArgumentException("Too big!");
+        }
+
+        final int capacity = (int) size;
+        if (array.length >= capacity) {
+            return;
+        }
+
+        // Grow to the nearest bigger power of 2
+        final int newSize = Integer.highestOneBit(capacity) << 1;
+        array = new byte[newSize];
+        segment = MemorySegment.ofArray(array);
+    }
+
+    interface ArrayConsumer {
+        void process(byte[] array) throws IOException;
+    }
+}

src/main/java/ru/vk/itmo/test/tarazanovmaxim/dao/LiveFilteringIterator.java

@@ -0,0 +1,52 @@
+package ru.vk.itmo.test.tarazanovmaxim.dao;
+
+import ru.vk.itmo.dao.Entry;
+
+import java.lang.foreign.MemorySegment;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+/**
+ * Filters non tombstone {@link Entry}s.
+ *
+ * @author incubos
+ */
+final class LiveFilteringIterator implements Iterator<Entry<MemorySegment>> {
+    private final Iterator<Entry<MemorySegment>> delegate;
+    private Entry<MemorySegment> next;
+
+    LiveFilteringIterator(final Iterator<Entry<MemorySegment>> delegate) {
+        this.delegate = delegate;
+        skipTombstones();
+    }
+
+    private void skipTombstones() {
+        while (delegate.hasNext()) {
+            final Entry<MemorySegment> entry = delegate.next();
+            if (entry.value() != null) {
+                this.next = entry;
+                break;
+            }
+        }
+    }
+
+    @Override
+    public boolean hasNext() {
+        return next != null;
+    }
+
+    @Override
+    public Entry<MemorySegment> next() {
+        if (!hasNext()) {
+            throw new NoSuchElementException();
+        }
+
+        // Consume
+        final Entry<MemorySegment> result = next;
+        next = null;
+
+        skipTombstones();
+
+        return result;
+    }
+}