Пронин Валентин / ИТМО DWS / Stage 1

src/main/java/ru/vk/itmo/test/proninvalentin/MemorySegmentFactory.java

@@ -0,0 +1,44 @@
+package ru.vk.itmo.test.proninvalentin;
+
+import ru.vk.itmo.dao.BaseEntry;
+import ru.vk.itmo.dao.Entry;
+
+import java.lang.foreign.MemorySegment;
+import java.lang.foreign.ValueLayout;
+import java.nio.charset.StandardCharsets;
+
+public class MemorySegmentFactory {
+    public byte[] toByteArray(MemorySegment data) {
+        if (data == null) {
+            throw new IllegalArgumentException();
+        }
+
+        return data.toArray(ValueLayout.JAVA_BYTE);
+    }
+
+    public MemorySegment fromString(String data) {
+        return data == null
+                ? null
+                : MemorySegment.ofArray(data.getBytes(StandardCharsets.UTF_8));
+    }
+
+    public Entry<MemorySegment> toMemorySegment(String key, String value) {
+        if (key == null || value == null) {
+            throw new IllegalArgumentException();
+        }
+
+        MemorySegment msKey = fromString(key);
+        MemorySegment msValue = fromString(value);
+        return new BaseEntry<>(msKey, msValue);
+    }
+
+    public Entry<MemorySegment> toDeletedMemorySegment(String key) {
+        if (key == null) {
+            throw new IllegalArgumentException();
+        }
+
+        MemorySegment msKey = fromString(key);
+        return new BaseEntry<>(msKey, null);
+    }
+}
+

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

@@ -0,0 +1,143 @@
+package ru.vk.itmo.test.proninvalentin;
+
+import one.nio.http.HttpException;
+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.Dao;
+import ru.vk.itmo.dao.Entry;
+import ru.vk.itmo.test.reference.dao.ReferenceDao;
+
+import java.io.IOException;
+import java.lang.foreign.MemorySegment;
+import java.util.Set;
+
+public class Server extends HttpServer {
+    private final Dao<MemorySegment, Entry<MemorySegment>> dao;
+    private final MemorySegmentFactory msFactory;
+    private static final Logger logger = LoggerFactory.getLogger(Server.class);
+    private static final Set<Integer> SUPPORTED_HTTP_METHODS = Set.of(
+            Request.METHOD_GET,
+            Request.METHOD_PUT,
+            Request.METHOD_DELETE
+    );
+
+    public Server(ServiceConfig config, ReferenceDao dao) throws IOException {
+        super(createServerConfig(config));
+
+        this.dao = dao;
+        this.msFactory = new MemorySegmentFactory();
+    }
+
+    private static HttpServerConfig createServerConfig(ServiceConfig serviceConfig) {
+        HttpServerConfig serverConfig = new HttpServerConfig();
+        AcceptorConfig acceptorConfig = new AcceptorConfig();
+        acceptorConfig.port = serviceConfig.selfPort();
+        acceptorConfig.reusePort = true;
+
+        serverConfig.acceptors = new AcceptorConfig[]{acceptorConfig};
+        serverConfig.closeSessions = true;
+        return serverConfig;
+    }
+
+    @Override
+    public void handleDefault(Request request, HttpSession session) throws IOException {
+        int httpMethod = request.getMethod();
+        Response response = isSupportedMethod(httpMethod)
+                ? new Response(Response.BAD_REQUEST, Response.EMPTY)
+                : new Response(Response.METHOD_NOT_ALLOWED, Response.EMPTY);
+        session.sendResponse(response);
+    }
+
+    private boolean isSupportedMethod(int httpMethod) {
+        return SUPPORTED_HTTP_METHODS.contains(httpMethod);
+    }
+
+    @Override
+    public void handleRequest(Request request, HttpSession session) {
+        try {
+            super.handleRequest(request, session);
+        } catch (Exception e) {
+            logger.error("Error while processing request", e);
+
+            String responseCode = e.getClass() == HttpException.class
+                    ? Response.BAD_REQUEST
+                    : Response.INTERNAL_ERROR;
+            sendResponse(session, new Response(responseCode, Response.EMPTY));
+        }
+    }
+
+    private static void sendResponse(HttpSession session, Response response) {
+        try {
+            session.sendResponse(response);
+        } catch (IOException e) {
+            logger.error("Error while sending response", e);
+        }
+    }
+
+    @Override
+    public synchronized void stop() {
+        super.stop();
+    }
+
+    // region Handlers
+
+    @Path("/v0/entity")
+    @RequestMethod(Request.METHOD_PUT)
+    public Response upsert(@Param(value = "id", required = true) String id, Request request) {
+        if (isNullOrEmpty(id) || request.getBody() == null) {
+            return new Response(Response.BAD_REQUEST, Response.EMPTY);
+        }
+
+        MemorySegment key = msFactory.fromString(id);
+        MemorySegment value = MemorySegment.ofArray(request.getBody());
+        dao.upsert(new BaseEntry<>(key, value));
+        return new Response(Response.CREATED, Response.EMPTY);
+    }
+
+    @Path("/v0/entity")
+    @RequestMethod(Request.METHOD_GET)
+    public Response get(@Param(required = true, value = "id") String id) {
+        if (isNullOrEmpty(id)) {
+            return new Response(Response.BAD_REQUEST, Response.EMPTY);
+        }
+
+        MemorySegment key = msFactory.fromString(id);
+
+        Entry<MemorySegment> entry = dao.get(key);
+        if (entry == null) {
+            return new Response(Response.NOT_FOUND, Response.EMPTY);
+        }
+
+        byte[] value = msFactory.toByteArray(entry.value());
+        return Response.ok(value);
+    }
+
+    @Path("/v0/entity")
+    @RequestMethod(Request.METHOD_DELETE)
+    public Response delete(@Param(required = true, value = "id") String id) {
+        if (isNullOrEmpty(id)) {
+            return new Response(Response.BAD_REQUEST, Response.EMPTY);
+        }
+
+        Entry<MemorySegment> deletedMemorySegment = msFactory.toDeletedMemorySegment(id);
+        dao.upsert(deletedMemorySegment);
+        return new Response(Response.ACCEPTED, Response.EMPTY);
+    }
+
+    private boolean isNullOrEmpty(String str) {
+        return str == null || str.isEmpty();
+    }
+
+    // endregion
+}

src/main/java/ru/vk/itmo/test/proninvalentin/ServiceImpl.java

@@ -0,0 +1,51 @@
+package ru.vk.itmo.test.proninvalentin;
+
+import ru.vk.itmo.Service;
+import ru.vk.itmo.ServiceConfig;
+import ru.vk.itmo.dao.Config;
+import ru.vk.itmo.test.ServiceFactory;
+import ru.vk.itmo.test.reference.dao.ReferenceDao;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.concurrent.CompletableFuture;
+
+public class ServiceImpl implements Service {
+    private final Config daoConfig;
+    private final ServiceConfig config;
+    private ReferenceDao dao;
+    private Server server;
+
+    public ServiceImpl(ServiceConfig config) throws IOException {
+        int flushThresholdBytes = 1 << 27; // 128 MB
+        this.config = config;
+        this.daoConfig = new Config(config.workingDir(), flushThresholdBytes);
+    }
+
+    @Override
+    public CompletableFuture<Void> start() throws IOException {
+        dao = new ReferenceDao(daoConfig);
+        server = new Server(config, dao);
+        server.start();
+        return CompletableFuture.completedFuture(null);
+    }
+
+    @Override
+    public CompletableFuture<Void> stop() throws IOException {
+        server.stop();
+        dao.close();
+        return CompletableFuture.completedFuture(null);
+    }
+
+    @ServiceFactory(stage = 1)
+    public static class Factory implements ServiceFactory.Factory {
+        @Override
+        public Service create(ServiceConfig config) {
+            try {
+                return new ServiceImpl(config);
+            } catch (IOException e) {
+                throw new UncheckedIOException(e);
+            }
+        }
+    }
+}

src/main/java/ru/vk/itmo/test/proninvalentin/StartServer.java

@@ -0,0 +1,34 @@
+package ru.vk.itmo.test.proninvalentin;
+
+import ru.vk.itmo.ServiceConfig;
+import ru.vk.itmo.dao.Config;
+import ru.vk.itmo.test.reference.dao.ReferenceDao;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+
+public final class StartServer {
+    private StartServer() {
+        // Suppress warning
+    }
+
+    public static void main(String[] args) throws IOException {
+        String url = "http://localhost";
+        int port = 8080;
+        int flushThresholdBytes = 1 << 27; // 128 MB
+        Path profilingDataPath = Path.of(
+                "/Users/valentinpronin/IdeaProjects/2024-highload-dht/"
+                        + "src/main/java/ru/vk/itmo/test/"
+                        + "proninvalentin/server_profiling_data");
+        Files.createDirectories(profilingDataPath);
+
+        Config daoConfig = new Config(profilingDataPath, flushThresholdBytes);
+        ServiceConfig serviceConfig = new ServiceConfig(port, url, List.of(url), profilingDataPath);
+        ReferenceDao referenceDao = new ReferenceDao(daoConfig);
+
+        Server server = new Server(serviceConfig, referenceDao);
+        server.start();
+    }
+}

src/main/java/ru/vk/itmo/test/proninvalentin/lua_scripts/get.lua

@@ -0,0 +1,6 @@
+function request()
+    id = math.random(1, 300000)
+    path = "/v0/entity?id=" .. id
+    method = "GET"
+    return wrk.format(method, path, body)
+end

src/main/java/ru/vk/itmo/test/proninvalentin/lua_scripts/put.lua

@@ -0,0 +1,9 @@
+id = 0
+
+function request()
+    body = string.rep("random_value", 100)
+    method = "PUT"
+    id = id + 1
+    path = "/v0/entity?id=" .. id
+    return wrk.format(method, path, headers, body)
+end