Add ability to limit the total number of upload slots, as well as per user

core/src/main/groovy/com/muwire/core/Core.groovy

@@ -276,7 +276,7 @@ public class Core {
         eventBus.register(UIDownloadResumedEvent.class, downloadManager)
 
         log.info("initializing upload manager")
-        uploadManager = new UploadManager(eventBus, fileManager, meshManager, downloadManager)
+        uploadManager = new UploadManager(eventBus, fileManager, meshManager, downloadManager, props)
 
         log.info("initializing connection establisher")
         connectionEstablisher = new ConnectionEstablisher(eventBus, i2pConnector, props, connectionManager, hostCache)

core/src/main/groovy/com/muwire/core/MuWireSettings.groovy

@@ -17,6 +17,8 @@ class MuWireSettings {
     int trustListInterval
     Set<Persona> trustSubscriptions
     int downloadRetryInterval
+    int totalUploadSlots
+    int uploadSlotsPerUser
     int updateCheckInterval
     boolean autoDownloadUpdate
     String updateType
@@ -73,6 +75,8 @@ class MuWireSettings {
         searchComments = Boolean.valueOf(props.getProperty("searchComments","true"))
         browseFiles = Boolean.valueOf(props.getProperty("browseFiles","true"))
         speedSmoothSeconds = Integer.valueOf(props.getProperty("speedSmoothSeconds","60"))
+        totalUploadSlots = Integer.valueOf(props.getProperty("totalUploadSlots","-1"))
+        uploadSlotsPerUser = Integer.valueOf(props.getProperty("uploadSlotsPerUser","-1"))
 
         watchedDirectories = readEncodedSet(props, "watchedDirectories")
         watchedKeywords = readEncodedSet(props, "watchedKeywords")
@@ -118,6 +122,8 @@ class MuWireSettings {
         props.setProperty("searchComments", String.valueOf(searchComments))
         props.setProperty("browseFiles", String.valueOf(browseFiles))
         props.setProperty("speedSmoothSeconds", String.valueOf(speedSmoothSeconds))
+        props.setProperty("totalUploadSlots", String.valueOf(totalUploadSlots))
+        props.setProperty("uploadSlotsPerUser", String.valueOf(uploadSlotsPerUser))
 
         writeEncodedSet(watchedDirectories, "watchedDirectories", props)
         writeEncodedSet(watchedKeywords, "watchedKeywords", props)

core/src/main/groovy/com/muwire/core/upload/UploadManager.groovy

@@ -4,6 +4,8 @@ import java.nio.charset.StandardCharsets
 
 import com.muwire.core.EventBus
 import com.muwire.core.InfoHash
+import com.muwire.core.MuWireSettings
+import com.muwire.core.Persona
 import com.muwire.core.SharedFile
 import com.muwire.core.connection.Endpoint
 import com.muwire.core.download.DownloadManager
@@ -22,15 +24,22 @@ public class UploadManager {
     private final FileManager fileManager
     private final MeshManager meshManager
     private final DownloadManager downloadManager
+    private final MuWireSettings props
 
+    /** LOCKING: this on both structures */
+    private int totalUploads
+    private final Map<Persona, Integer> uploadsPerUser = new HashMap<>()
+    
     public UploadManager() {}
 
     public UploadManager(EventBus eventBus, FileManager fileManager,
-        MeshManager meshManager, DownloadManager downloadManager) {
+        MeshManager meshManager, DownloadManager downloadManager,
+        MuWireSettings props) {
         this.eventBus = eventBus
         this.fileManager = fileManager
         this.meshManager = meshManager
         this.downloadManager = downloadManager
+        this.props = props
     }
 
     public void processGET(Endpoint e) throws IOException {
@@ -82,7 +91,15 @@ public class UploadManager {
 
             if (request.have > 0)
                 eventBus.publish(new SourceDiscoveredEvent(infoHash : request.infoHash, source : request.downloader))
-
+                
+            if (!incrementUploads(request.downloader)) {
+                log.info("rejecting due to slot limit")
+                e.getOutputStream().write("429 Too Many Requests\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
+                e.getOutputStream().flush()
+                e.close()
+                return
+            }
+            
             Mesh mesh
             File file
             int pieceSize
@@ -103,6 +120,7 @@ public class UploadManager {
             try {
                 uploader.respond()
             } finally {
+                decrementUploads(request.downloader)
                 eventBus.publish(new UploadFinishedEvent(uploader : uploader))
             }
         }
@@ -157,12 +175,21 @@ public class UploadManager {
                 return
             }
         }
+        
+        if (!incrementUploads(request.downloader)) {
+            log.info("rejecting due to slot limit")
+            e.getOutputStream().write("429 Too Many Requests\r\n\r\n".getBytes(StandardCharsets.US_ASCII))
+            e.getOutputStream().flush()
+            e.close()
+            return
+        }
 
         Uploader uploader = new HashListUploader(e, fullInfoHash, request)
         eventBus.publish(new UploadEvent(uploader : uploader))
         try {
             uploader.respond()
         } finally {
+            decrementUploads(request.downloader)
             eventBus.publish(new UploadFinishedEvent(uploader : uploader))
         }
 
@@ -233,5 +260,37 @@ public class UploadManager {
             }
         }
     }
+    
+    /**
+     * @param p downloader
+     * @return true if this upload hasn't hit any slot limits
+     */
+    private synchronized boolean incrementUploads(Persona p) {
+        if (props.totalUploadSlots >= 0 && totalUploads >= props.totalUploadSlots)
+            return false
+        if (props.uploadSlotsPerUser == 0)
+            return false
+        
+        Integer currentUploads = uploadsPerUser.get(p)
+        if (currentUploads == null)
+            currentUploads = 0
+        if (props.uploadSlotsPerUser > 0 && currentUploads >= props.uploadSlotsPerUser)
+            return false
+        uploadsPerUser.put(p, ++currentUploads)
+        totalUploads++
+        true
+    }
+    
+    private synchronized void decrementUploads(Persona p) {
+        totalUploads--
+        Integer currentUploads = uploadsPerUser.get(p)
+        if (currentUploads == null || currentUploads == 0)
+            throw new IllegalStateException()
+        currentUploads--
+        if (currentUploads == 0)
+            uploadsPerUser.remove(p)
+        else
+            uploadsPerUser.put(p, currentUploads)
+    }
 }