Zubakov. Torrent

Readme.md

@@ -0,0 +1,150 @@
+# Torrent
+
+* На трекере хранится список файлов и информация об активных пользователях, у которых есть те или иные файлы (возможно не целиком).
+
+* С помощью клиентского приложения можно просматривать список файлов на трекере, а также добавлять новые и выбирать файлы из списка для скачивания.
+
+* Файлы условно разбиваются на последовательные блоки бинарных данных константного размера (например 10M). Последний блок может иметь меньший размер. Блоки нумеруются с нуля.
+
+---
+
+# Torrent
+
+* Клиент при подключении отправляет на трекер список раздаваемых им файлов.
+
+* При скачивании файла клиент получает у трекера информацию о клиентах, раздающих файл (сидах), и далее общается с ними напрямую.
+
+* У отдельного сида можно узнать о том, какие полные части у него есть, а также скачать их.
+
+* После скачивания отдельных блоков некоторого файла клиент становится сидом.
+
+---
+
+# Torrent-tracker
+
+* Хранит мета-информацию о раздаваемых файлах:
+    * идентификатор
+    * активные клиенты (недавно был update), у которых есть этот файл целиком или некоторые его части
+
+* Порт сервера: 8081
+
+* Запросы:
+    * list — список раздаваемых файлов
+    * upload — публикация нового файла
+    * sources — список клиентов, владеющих определенным файлов целиком или некоторыми его частями
+    * update — загрузка клиентом данных о раздаваемых файлах 
+
+---
+
+# List
+
+Формат запроса:
+
+    <1: Byte>
+Формат ответа:
+
+    <count: Int> (<id: Int> <name: String> <size: Long>)*,
+    count — количество файлов
+    id — идентификатор файла
+    name — название файла
+    size — размер файла
+
+---
+
+# Upload
+
+Формат запроса:
+
+    <2: Byte> <name: String> <size: Long>,
+    name — название файла
+    size — размер файла
+Формат ответа:
+
+    <id: Int>,
+    id — идентификатор файла
+
+# Примечание
+
+* Если клиент А и клиент Б решили опубликовать файл abc.txt, то это будут **разные** файлы, иными словами каждый запрос на публикацию файла возвращает **новый** id
+
+---
+
+# Sources
+
+Формат запроса:
+
+    <3: Byte> <id: Int>,
+    id — идентификатор файла
+Формат ответа:
+
+    <size: Int> (<ip: ByteByteByteByte> <clientPort: Short>)*,
+    size — количество клиентов, раздающих файл
+    ip — ip клиента,
+    clientPort — порт клиента
+
+---
+
+# Update
+
+Формат запроса:
+
+    <4: Byte> <clientPort: Short> <count: Int> (<id: Int>)*,
+    clientPort — порт клиента,
+    count — количество раздаваемых файлов,
+    id — идентификатор файла
+Формат ответа:
+
+    <status: Boolean>,
+    status — True, если информация успешно обновлена
+
+# Примечание
+
+* Клиент обязан исполнять данный запрос каждые 5 минут, иначе сервер считает, что клиент ушел с раздачи
+
+---
+
+# Torrent-client
+
+* Порт клиента указывается при запуске и передается на трекер в рамках запроса update
+
+* Каждый файл раздается по частям, размер части — константа на всё приложение
+
+* Клиент хранит и раздает эти самые части
+
+* Запросы:
+    * stat — доступные для раздачи части определенного файла
+    * get — скачивание части определенного файла
+
+---
+
+# Stat
+
+Формат запроса:
+
+    <1: Byte> <id: Int>,
+    id — идентификатор файла
+Формат ответа:
+
+    <count: Int> (<part: Int>)*,
+    count — количество доступных частей
+    part — номер части
+
+# Примечание
+
+* Часть считается доступной для раздачи, если она хранится на клиенте целиком
+
+---
+
+# Get
+
+Формат запроса:
+
+    <2: Byte> <id: Int> <part: Int>
+    id — идентификатор файла,
+    part — номер части 
+Формат ответа:
+
+    <content: Bytes>,
+    content — содержимое части
+
+---

build.gradle

@@ -0,0 +1,20 @@
+group 'org.itmo'
+version '1.0-SNAPSHOT'
+
+apply plugin: 'java'
+
+sourceCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    compile project(':tracker')
+    compile project(':client')
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+}
+
+task wrapper(type: Wrapper) {
+    gradleVersion = '4.3'
+}

client/build.gradle

@@ -0,0 +1,28 @@
+group 'org.itmo'
+version '1.0-SNAPSHOT'
+
+apply plugin: 'java'
+
+sourceCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+jar {
+    from {
+        configurations.compile.collect {
+            it.isDirectory() ? it : zipTree(it)
+        }
+    }
+    manifest {
+        attributes 'Main-Class': 'org.itmo.torrent.client.TorrentClientMain'
+    }
+}
+
+dependencies {
+    compile project(':commons')
+    compile group: 'commons-io', name: 'commons-io', version: '2.6'
+    compile group: 'org.jetbrains', name: 'annotations', version: '15.0'
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+}

client/src/main/java/org/itmo/torrent/client/Client.java

@@ -0,0 +1,35 @@
+package org.itmo.torrent.client;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.itmo.torrent.filesystem.FileInfo;
+import org.itmo.torrent.network.Address;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.List;
+import java.util.Set;
+
+public interface Client {
+
+    @NotNull
+    List<FileInfo> executeList() throws IOException;
+
+    int executeUpload(@NotNull final File file) throws IOException;
+
+    @NotNull
+    Set<Address> executeSources(final int fileId) throws IOException;
+
+    boolean executeUpdate() throws IOException;
+
+    void executeDownload(final int fileId) throws IOException;
+
+    @NotNull
+    Set<Integer> executeStat(@NotNull final Address address,
+                             final int fileId) throws IOException;
+
+    @Nullable
+    InputStream executeGet(@NotNull final Address address,
+                           final int fileId, final int fileChunkId) throws IOException;
+}

client/src/main/java/org/itmo/torrent/client/Downloader.java

@@ -0,0 +1,159 @@
+package org.itmo.torrent.client;
+
+import org.apache.commons.io.IOUtils;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import org.itmo.torrent.filesystem.FileInfo;
+import org.itmo.torrent.filesystem.FileSeedInfo;
+import org.itmo.torrent.network.Address;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.RandomAccessFile;
+import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.RecursiveTask;
+
+public class Downloader implements AutoCloseable {
+    @NotNull
+    private final TorrentClient client;
+    @NotNull
+    private final ExecutorService executor = Executors.newWorkStealingPool();
+
+    public Downloader(@NotNull final TorrentClient client) {
+        this.client = client;
+    }
+
+    void download(final int fileId) {
+        executor.submit(new DownloadFileTask(fileId));
+    }
+
+    @Override
+    public void close() {
+        executor.shutdown();
+    }
+
+    private class DownloadFileTask implements Runnable {
+        private final int fileId;
+
+        DownloadFileTask(final int fileId) {
+            this.fileId = fileId;
+        }
+
+        @Override
+        public void run() {
+            final FileSeedInfo fileInfo = getFileSeedInfo();
+            if (fileInfo == null) {
+                return;
+            }
+
+            final Map<Integer, Set<Address>> fileChunkSeeders = getFileChunksSeeders();
+            if (fileChunkSeeders == null) {
+                return;
+            }
+
+            try (RandomAccessFile out = new RandomAccessFile(fileInfo.getFile(), "rw")) {
+                out.setLength(fileInfo.getSize());
+                client.getState().addFileInfo(fileInfo);
+                final List<DownloadFileChunkTask> subTasks = fork(fileInfo, fileChunkSeeders);
+                join(fileInfo, out, subTasks);
+            } catch (IOException ignore) {
+                // TODO: proper handling
+            }
+        }
+
+        @Nullable
+        private Map<Integer, Set<Address>> getFileChunksSeeders() {
+            // TODO: fix bottleneck
+            final Map<Integer, Set<Address>> fileChunkSeeders = new HashMap<>();
+            try {
+                for (Address source : client.executeSources(fileId)) {
+                    for (int chunksId : client.executeStat(source, fileId)) {
+                        fileChunkSeeders.putIfAbsent(chunksId, new HashSet<>());
+                        fileChunkSeeders.get(chunksId).add(source);
+                    }
+                }
+            } catch (IOException e) {
+                System.err.println(e.getMessage());
+                return null;
+            }
+
+            return fileChunkSeeders;
+        }
+
+        @NotNull
+        private List<DownloadFileChunkTask> fork(@NotNull final FileSeedInfo fileInfo,
+                                                 @NotNull final Map<Integer, Set<Address>> fileChunkSeeders) {
+            return new ArrayList<DownloadFileChunkTask>() {{
+                for (int fileChunkId = 0; fileChunkId < fileInfo.getChunksNumber(); fileChunkId++) {
+                    if (!fileInfo.isChunkAvailable(fileChunkId)) {
+                        final DownloadFileChunkTask task = new DownloadFileChunkTask(fileId, fileChunkId,
+                                fileChunkSeeders.getOrDefault(fileChunkId, Collections.emptySet()));
+                        task.fork();
+                        add(task);
+                    }
+                }
+            }};
+        }
+
+        private void join(@NotNull final FileSeedInfo fileInfo, @NotNull final RandomAccessFile out,
+                          @NotNull final List<DownloadFileChunkTask> subTasks) throws IOException {
+            for (DownloadFileChunkTask task : subTasks) {
+                final InputStream fileChunkContent = task.join();
+                out.seek(task.fileChunkId * FileInfo.FILE_CHUNK_SIZE);
+                out.write(IOUtils.toByteArray(fileChunkContent));
+                fileInfo.setChunkAvailable(task.fileChunkId);
+            }
+        }
+
+        @Nullable
+        private FileSeedInfo getFileSeedInfo() {
+            FileSeedInfo fileInfo = client.getState().getFileInfo(fileId);
+            if (fileInfo == null) {
+                try {
+                    FileInfo tempFileInfo = client.executeList().stream()
+                            .filter(info -> info.getId() == fileId)
+                            .findAny()
+                            .orElse(null);
+                    fileInfo = tempFileInfo != null ? new FileSeedInfo(tempFileInfo) : null;
+                } catch (IOException ignored) {
+                }
+            }
+
+            return fileInfo;
+        }
+    }
+
+    private class DownloadFileChunkTask extends RecursiveTask<InputStream> {
+        private final int fileId;
+        private final int fileChunkId;
+        @NotNull
+        private final Set<Address> seeders;
+
+        DownloadFileChunkTask(final int fileId, final int fileChunkId,
+                              @NotNull final Set<Address> seeders) {
+            this.fileId = fileId;
+            this.fileChunkId = fileChunkId;
+            this.seeders = seeders;
+        }
+
+        @Nullable
+        @Override
+        protected InputStream compute() {
+            for (Address address : seeders) {
+                InputStream fileChunkContent = null;
+                try {
+                    fileChunkContent = client.executeGet(address, fileId, fileChunkId);
+                } catch (IOException ignored) {
+                }
+                if (fileChunkContent != null) {
+                    return fileChunkContent;
+                }
+            }
+
+            return null;
+        }
+    }
+}
+