@@ -33,6 +33,12 @@ max_memory_size = 134217728 # then if the journal grows larger than 160MB, it will be forcably re-created. max_journal_overflow = 10 +# if you would like a timer to periodically sweep through queues and clean +# up expired items (when they are at the head of a queue), set the timer's +# frequency here. 0 (the default) turns it off. this is only useful for queues +# that are rarely (or never) polled, but may contain short-lived items. +#expiration_timer_frequency_seconds = 1 + # per-queue config queues { weather_updates { @@ -40,7 +46,7 @@ queues { # seconds (1,800,000 milliseconds, or 30 mins). if a client uses a # shorter expiry, that's honored instead. max_age = 1800000 - + # refuse SET operations when the queue would exceed this many items. max_items = 1500000 } config/production.conf @@ -2,7 +2,7 @@ <info organisation="net.lag" module="kestrel" - revision="1.2" + revision="1.2.1" e:testclass="net.lag.TestRunner" e:jarclassname="net.lag.kestrel.Kestrel" e:buildpackage="net.lag.kestrel" ivy/ivy.xml @@ -19,6 +19,7 @@ package net.lag.kestrel import java.net.InetSocketAddress import java.util.concurrent.{CountDownLatch, Executors, ExecutorService, TimeUnit} +import java.util.{Timer, TimerTask} import scala.actors.{Actor, Scheduler} import scala.actors.Actor._ import scala.collection.mutable @@ -108,6 +109,21 @@ object Kestrel { // expose config thru JMX. config.registerWithJmx("net.lag.kestrel") + // optionally, start a periodic timer to clean out expired items. + val expirationTimerFrequency = config.getInt("expiration_timer_frequency_seconds", 0) + if (expirationTimerFrequency > 0) { + val timer = new Timer("Expiration timer", true) + val expirationTask = new TimerTask { + def run() { + val expired = Kestrel.queues.flushAllExpired() + if (expired > 0) { + log.info("Expired %d item(s) from queues automatically.", expired) + } + } + } + timer.schedule(expirationTask, expirationTimerFrequency * 1000, expirationTimerFrequency * 1000) + } + log.info("Kestrel started.") // make sure there's always one actor running so scala 2.7.2 doesn't kill off the actors library. src/main/scala/net/lag/kestrel/Kestrel.scala @@ -137,7 +137,7 @@ class KestrelHandler(val session: IoSession, val config: Config) extends Actor { case "FLUSH_EXPIRED" => flushExpired(request.line(1)) case "FLUSH_ALL_EXPIRED" => - val flushed = Kestrel.queues.queueNames.foldLeft(0) { (sum, qName) => sum + Kestrel.queues.flushExpired(qName) } + val flushed = Kestrel.queues.flushAllExpired() writeResponse("%d\r\n".format(flushed)) case "VERSION" => version() src/main/scala/net/lag/kestrel/KestrelHandler.scala @@ -200,6 +200,10 @@ class QueueCollection(queueFolder: String, private var queueConfigs: ConfigMap) } } + def flushAllExpired(): Int = synchronized { + queueNames.foldLeft(0) { (sum, qName) => sum + flushExpired(qName) } + } + def stats(key: String): Array[(String, String)] = queue(key) match { case None => Array[(String, String)]() case Some(q) => src/main/scala/net/lag/kestrel/QueueCollection.scala 0d187c741bbbf60d3cc84fba2f5b497ae8733350 Robey Pointer robey@twitter.com http://github.com/robey/kestrel/commit/310d5fb7b0f92ea6c24798978a1182fdbdd674c1 310d5fb7b0f92ea6c24798978a1182fdbdd674c1 2009-10-09T18:45:12-07:00 2009-10-09T18:45:12-07:00 add a config option to periodically sweep through queues and clear out expired items. 2d89c42b5f58a7ba7178ab4763e74c4a64e05aa1 Robey Pointer robey@twitter.com