[입 개발] Redis internal : Redis Dictionary Type 에 관해서

Redis 는 기본적으로 Hash 형태로 데이터를 관리하지만, 내부적으로 관리가 필요한 정보들 역시, 내부적으로는 Dictionary 라고 부르는 Hash 형태로 관리하고 있습니다. 그런데 이 Dictionary 에 대한 핸들링이, 관리되는 데이터의 종류에 따라서 다르게 처리되어야 할 때가 있습니다. 아마 오늘 글을 그냥 Redis를 쓰시는 분 입장에서는 별 내용이 없고, Redis 소스를 건드리시는 분들에게는 아주 살짝 도움이 될듯합니다.

보통 대부분의 언어에서는 함수 오버라이딩등을 이용한 폴리모피즘을 이용해서 이런 방식의 요구사항을 좀 수월하게 처리하게 되어 있습니다. 그러나 C 에서는… 기본적으로 이런 방식이 제공되지 않지만, 함수 포인터를 이용해서 이런 방식을 구현할 수 있습니다. 가장 유명한 예가, 리눅스 커널의 VFS 같은 부분을 보면 됩니다.

cluster.c 의 clusterInit 코드를 보면 다음과 같이 dictCreate 함수를 이용해서 dictionary를 생성하는 것을 볼 수 있습니다.

    server.cluster->nodes = dictCreate(&clusterNodesDictType,NULL);
    server.cluster->nodes_black_list =
        dictCreate(&clusterNodesBlackListDictType,NULL);

dictCreate 함수를 살펴보면 다음과 같이 dictType, privDataPtr 두개의 파라매터를 가지고, 결과로 dict 를 넘겨줍니다.

dict *dictCreate(dictType *type,
        void *privDataPtr)
{
    dict *d = zmalloc(sizeof(*d));

    _dictInit(d,type,privDataPtr);
    return d;
}

먼저 dict 구조체부터 살펴보도록 하겠습니다.

typedef struct dict {
    dictType *type;
    void *privdata;
    dictht ht[2];
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    int iterators; /* number of iterators currently running */
} dict;

dict 구조체는 dictType 과 dictht 두개를 가집니다. 여기서 dictht는 dict의 테이블 확장시에 사용하기 위한 것입니다. 그럼 이 dict 안에 있는 dictType은 뭘까요? dictType은 다음과 같이 정의되어 있습니다.

typedef struct dictType {
    unsigned int (*hashFunction)(const void *key);
    void *(*keyDup)(void *privdata, const void *key);
    void *(*valDup)(void *privdata, const void *obj);
    int (*keyCompare)(void *privdata, const void *key1, const void *key2);
    void (*keyDestructor)(void *privdata, void *key);
    void (*valDestructor)(void *privdata, void *obj);
} dictType;

dictType 을 보시면 hashFuction, keyDup, valDup, keyCompare, keyDestructor, valDestructor 같은 값들이 정의되어 있습니다. 즉, 함수포인터를 가지고 있기 때문에, 여기서 가리키는 함수의 동작이 다르면, 같은 형태로 보이더라도 서로 다르게 동작하게 할 수 있습니다.

그리고 현재 Redis 에는 다음과 같이 8개의 dictType이 정의되어 있습니다.

  • dictType setDictType;
  • dictType clusterNodesDictType;
  • dictType clusterNodesBlackListDictType;
  • dictType dbDictType;
  • dictType shaScriptObjectDictType;
  • dictType hashDictType;
  • dictType replScriptCacheDictType;

dictType 구조체 안에서 특히 keyCompare 를 통해서 해당 키를 찾아내게 됩니다.
예를 들어서, 내부적으로 실제 키들을 저장하는 곳에서는 dbDictType 을 사용합니다. dbDictType을 보면 다음과 같습니다.

/* Db->dict, keys are sds strings, vals are Redis objects. */
dictType dbDictType = {
    dictSdsHash,                /* hash function */
    NULL,                       /* key dup */
    NULL,                       /* val dup */
    dictSdsKeyCompare,          /* key compare */
    dictSdsDestructor,          /* key destructor */
    dictRedisObjectDestructor   /* val destructor */
};

dbDictType에서는 Key의 비교를 위해서는 dictSdsKeyCompare 를 사용하고, dictSdsKeyCompare는 다음과 같이 구현되어 있습니다.

int dictSdsKeyCompare(void *privdata, const void *key1,
        const void *key2)
{
    int l1,l2;
    DICT_NOTUSED(privdata);

    l1 = sdslen((sds)key1);
    l2 = sdslen((sds)key2);
    if (l1 != l2) return 0;
    return memcmp(key1, key2, l1) == 0;
}

그리고 Redis Command 를 저장하는 곳에서 사용하는 commandTableDictType 에서는 대소문자 구분이 필요없을 때는 dictSdsKeyCaseCompare 를 사용합니다.

int dictSdsKeyCaseCompare(void *privdata, const void *key1,
        const void *key2)
{
    DICT_NOTUSED(privdata);

    return strcasecmp(key1, key2) == 0;
}

위와 같은 형태에서 dictFind를 살펴보면 내부적으로 dictType 함수포인터를 이용하게 됩니다.

dictEntry *dictFind(dict *d, const void *key)
{
    dictEntry *he;
    unsigned int h, idx, table;

    if (d->ht[0].size == 0) return NULL; /* We don't have a table at all */
    if (dictIsRehashing(d)) _dictRehashStep(d);
    h = dictHashKey(d, key);
    for (table = 0; table <= 1; table++) {
        idx = h & d->ht[table].sizemask;
        he = d->ht[table].table[idx];
        while(he) {
            if (dictCompareKeys(d, key, he->key))
                return he;
            he = he->next;
        }
        if (!dictIsRehashing(d)) return NULL;
    }
    return NULL;
}

#define dictCompareKeys(d, key1, key2) \
    (((d)->type->keyCompare) ? \
        (d)->type->keyCompare((d)->privdata, key1, key2) : \
        (key1) == (key2))

위의 dictCompareKeys가 내부적으로 dictType의 keyCompare를 이용하는 것을 볼 수 있습니다. 그런데 여기서 조심해야 하는 것들이 있습니다. 실제 dictFind 에서 key를 넘겨줄 때, dictSdsKeyCompare 는 key가 무조건 sds type이어야 하지만, dictSdsKeyCaseCompare 에서는 실제 값만 비교하므로, 단순 string 이어도 가능합니다. 말 그대로 가능만 합니다. 그런데, 뭔가 잘못 쓰기 시작하면… Redis가 뻥뻥 죽어나가는 걸 보실 수 있을껍니다.

그래서 cluster.c 소스를 보면, 무조건 key가 sds 형태여야 하기 때문에 발생하는 사소한 오버헤드도 존재합니다. 뭐, 그러나 자주 발생하지는 않으니… 그냥 패스를 ㅎㅎㅎ

[입 개발] Scala의 거시기 : _(underscore) 의 용법 정리

Scala 를 사용하면 만나게 되는 여러 문법 중에, 처음 접하는 이들의 머리를 깨는 것이 있으니, 그게 바로 Scala에서 거시기로 통하는 _(underscore) 입니다. 이게 뭐지 하고 고민하는 중에 팀분이(멀린 사랑해요.) 아래 자료를 알려주셨습니다. 그리고 이 블로그는 아래의 문서를 풀이하는 것입니다. 사실 아래의 문서만 봐도 거의 모든 것이 이해되지만, 제 부족한 머리를 위해서 정리해둡니다. 참고로 dreaded 는 “무서운” 이런뜻입니다.

첫 슬라이드는 다음과 같습니다. 뭔가 어려워보이죠? 각종 _ 의 용법은 모두 들어가 있습니다.
아주 간단하게 설명하면 아래의 offset 에 대입되는 커링 함수 sum2(count) 에서 count는 매번 실행시의 count 값이 바인딩됩니다.(악 벌써 여기부터 어려워!!!)

class Underscores {
	import collection.{ Map => _ , _ }

	var count : Int = _

	def sum = (_:Int) + (_:Int)
	def sum2(a:Int)(b:Int) = a+b
	def offset = sum2(count) _

	def sizeOf(l:Traversable[_]) : Unit = l match {
		case it:Iterable[Int] => count = (0/:it)(_ + _)
		case s:Seq[_] => s.foreach( _ => count = count + 1)
		case _ => println(offset(l.size))
	}
}

두번째 슬라이드는 각각 어떤 용법으로 이루어졌나를 보여줍니다. 총 6가지 용법을 색상까지 이쁘게 보여주네요.

세번째 슬라이드 부터 각각의 용법에 대해서 알려줍니다.

1번은 “모두”를 의미합니다. 아래의 예에서 첫번째 Map => _ 는 일단 무시하시고(5번이니깐요.)
그 뒤의 _ 가 자바에서의 import * 와 같은 용법입니다.(왜 *가 아닌지는…)

	import collection.{ Map => _ , _ }

실제 예는 다음과 같습니다.

import java.util._
val date = new Date()

import scala.util.control.Breaks._
breakable {
	for (i <- 0 to 10) { if (i == 5) break }
}

웬지 breakable은 Exception을 잡아서 나가는 것 같은 느낌이 진하게 납니다.

2번은 디폴트 값 지정입니다. 숫자면 0 문자면 null로 지정됩니다.

	var count : Int = _

다만 이렇게 지정하는 건 생성자에서만 되고 함수안에서는 되지 않습니다.

class Foo {
	var i:Int = _ // i = 0
	var s:String = _ // s = null

	def f {
	// var i:String = _//error: local variables must be initialized
	}
}

3번째는 unused variables 입니다. 아래와 같이 _로 받은걸 쓰지 않게 되는 겁니다.

		case _ => println(offset(l.size))

예를 보면 다음과 같습니다. 아래의 두 예는 같은 예입니다. x를 파라매터로 받지만… 쓰지 않는 겁니다.

(1 to 5) foreach { (x:Int) => println("one more")}
(1 to 5) foreach { _ => println("one more")}

다음 예제들도 동일합니다.

def inPatternMatching1(s:String) {
	s match {
		case "foo" => println("foo !")
		case x => println("not foo")
	}
}

def inPatternMatching2(s:String) {
	s match {
		case "foo" => println("foo !")
		case _ => println("not foo")
	}
}

4번째는 이름 없는 파라매터입니다. 아주 명확하게 들어갈 변수 대신에 지정되게 됩니다.(3번하고는 반대입죠.)
아래의 예는 1…10 까지가 x로 바인딩되는데, 이걸 _로 대체하는 경우입니다. 명시적으로 사용하기 위해서이죠.

(1 to 10) map { x => x + 1}
(1 to 10) map { _ + 1}

(1 to 10).foldLeft(0) { (x,y) => x+y }
(1 to 10).foldLeft(0) { _+_ }

이제 partial function 에서 보면 더더욱 눈이 휘둥굴해 집니다.

def f(i:Int) : String = i.toString

def g = (x:Int) => f(x)
def g2 = f _
def f2 = (_:String).toString

def u(i:Int)(d:Double) : String = i.toString + d.toString

def v = u _

def w1 = u(4) _

def w2 = u(_:Int)2.0)

5번은 아까 Map 관련 헤더들을 import 하지 말라는 뜻입니다.
6번은 c++의 Template 처럼 특정 타입을 지칭하는 것입니다. 위에서 넘어온 타입을 그대로 사용하겠다라고 하면 이해가 쉬울까요.

다시 한번 말씀드리지만, 슬라이드가 잘 되어있으니, 실제로 보시면 꽤 도움이 되실껍니다. 뭐, 저도 공부하는 중이라…

[입 개발] Scala 의 App Trait는 어떻게 동작하는가?

요새 스칼라 스터디를 하고 있는데…(스칼라 어려워요 흑흑흑, 전 스맹 T.T) 아주 여러가지 기능들이 있습니다. 그런데 첫 부분에 나오는 예제부터 머리속을 땡땡 때리는 경우가 있습니다. 간단한 예를 들자면, tuple의 파라매터가 22개 밖에 안되는 것은 실제로 tuple1 ~ tuple22 까지의 클래스가 있어서 처리된다는 것(tuple 은 다시 product 이라는 것을 상속받는…)

보통 우리가 언어를 처음 배울때 쓰는 첫 예제는… 반가워 세상입니다. 즉 Hello World! 를 출력하는 것이죠.

 object HelloWorld {
    def main(args: Array[String]) {
      println("Hello, world!")
    }
  }

그런데 App 이라는 trait 를 상속받으면 다음과 같은 형태로 똑같이 동작이 됩니다.

object HelloWorld extends App {
  println("Hello, world!")
}

사실 스칼라를 공부하는 사람이야 그냥 당연하게 넘어갈 수 있지만, 두 번째 예의 경우는 println 코드가 있는 부분은 생성자입니다. 그런데 “어떻게 저게 자동으로 실행이 되는거지?” 라는 의문이 생기게 됩니다.(안생기면 500원…), 그리고 args 도 사용할 수 있습니다.

그래서 안을 조금 파보니…

App Trait 는 다시 DelayedInit 라는 Trait를 상속받습니다. 먼저 App Trait 부터 살짝 보도록 하겠습니다.

trait App extends DelayedInit {

  /** The time when the execution of this program started, in milliseconds since 1
    * January 1970 UTC. */
  @deprecatedOverriding("executionStart should not be overridden", "2.11.0")
  val executionStart: Long = currentTime

  /** The command line arguments passed to the application's `main` method.
   */
  @deprecatedOverriding("args should not be overridden", "2.11.0")
  protected def args: Array[String] = _args

  private var _args: Array[String] = _

  private val initCode = new ListBuffer[() => Unit]

  /** The init hook. This saves all initialization code for execution within `main`.
   *  This method is normally never called directly from user code.
   *  Instead it is called as compiler-generated code for those classes and objects
   *  (but not traits) that inherit from the `DelayedInit` trait and that do not
   *  themselves define a `delayedInit` method.
   *  @param body the initialization code to be stored for later execution
   */
  @deprecated("The delayedInit mechanism will disappear.", "2.11.0")
  override def delayedInit(body: => Unit) {
    initCode += (() => body)
  }

    /** The main method.
   *  This stores all arguments so that they can be retrieved with `args`
   *  and then executes all initialization code segments in the order in which
   *  they were passed to `delayedInit`.
   *  @param args the arguments passed to the main method
   */
  @deprecatedOverriding("main should not be overridden", "2.11.0")
  def main(args: Array[String]) = {
    this._args = args
    for (proc <- initCode) proc()
    if (util.Properties.propIsSet("scala.time")) {
      val total = currentTime - executionStart
      Console.println("[total " + total + "ms]")
    }
  }
}

젤 아래의 main을 보면, 아 여기서 실행되겠구나 할것입니다. 쉽네하고 보다보면, 다시 이상해집니다. 분명히 main이 보통 entrypoint 일텐데…(실제로는 object이니 이것을 실행하는 부분이 있긴하겠죠.) 뭔가 initCode 라는 것에서 proc를 가져와서 이걸 실행시킵니다.

그 위의 delayedInit 함수를 보니, body가 넘어와서 initCode에 저장됩니다.(여기서 body는 람다라고 보시면 될듯합니다.)

그럼 다시 처음으로 여기서 main이 실행되는 건 알겠는데… App Trait를 상속받은 object의 생성자를 실행을 시켜주는 걸로 봐서 아마도 위의 proc 가 App Trait를 상속받은 object의 생성자일꺼라는 예상을 할 수 있게 됩니다. 그러나, 여전히 delayedInit을 호출해주는 녀석은 보이지 않습니다. 다시 App Trait 가 DelayedInit Trait를 상속받으니, 이걸 살펴보도록 하겠습니다.

trait DelayedInit {
  def delayedInit(x: => Unit): Unit
}

악!!! 살펴볼 내용이 없습니다. 그냥 인터페이스만 정의가 되어있습니다. 그럼 뭔가 언어적으로 뭔가 해주지 않을까 싶습니다. 소스를 까보면 src/reflect/scala/reflect/internal/Definitions.scala 에서 다음 코드를 발견할 수 있습니다.

def delayedInitMethod = getMemberMethod(DelayedInitClass, nme.delayedInit)

해당 클래스에서 delayedInit를 뽑아내는 것 같습니다.

그리고 src/compiler/scala/tools/nsc/transform/Constructors.scala 를 보면 다음 코드가 있습니다.

    private def delayedInitCall(closure: Tree) = localTyper.typedPos(impl.pos) {
      gen.mkMethodCall(This(clazz), delayedInitMethod, Nil, List(New(closure.symbol.tpe, This(clazz))))
    }

그리고 위의 delayedInitCall은 rewriteDelayedInit() 에서 사용하고 있습니다. delayedInitCall을 실제로
호출하게 됩니다. 즉 여기서 아까 delayedInit가 호출되면서 App Trait 의 initCode 쪽에 생성자를 넣어주는 것입니다. 그래서 실제로 App Trait 의 main에서 그걸 호출하게 되는거죠.

    def rewriteDelayedInit() {
      /* XXX This is not corect: remainingConstrStats.nonEmpty excludes too much,
       * but excluding it includes too much.  The constructor sequence being mimicked
       * needs to be reproduced with total fidelity.
       *
       * See test case files/run/bug4680.scala, the output of which is wrong in many
       * particulars.
       */
      val needsDelayedInit = (isDelayedInitSubclass && remainingConstrStats.nonEmpty)

      if (needsDelayedInit) {
        val delayedHook: DefDef = delayedEndpointDef(remainingConstrStats)
        defBuf += delayedHook
        val hookCallerClass = {
          // transform to make the closure-class' default constructor assign the the outer instance to its pa>
          val drillDown = new ConstructorTransformer(unit)
          drillDown transform delayedInitClosure(delayedHook.symbol.asInstanceOf[MethodSymbol])
        }
        defBuf += hookCallerClass
        remainingConstrStats = delayedInitCall(hookCallerClass) :: Nil
      }
    }

마지막으로 Constructors.scala 안에서 다시 rewriteDelayedInit를 실행합니다. 그래서 App Trait를 상속받을 경우 생성자에만 코드를 넣어두면 실행이 되는 것입니다.

뭐, 이게 맞는 플로우인지는 정확하게 보증은 못합니다. 저도 이제 막 스칼라를 공부하는 중이고, 아무리 봐도, 스칼라를 편안하게 쓰지는 못할듯 하네요 T.T 흑흑흑

[입 개발] mosquitto build

mosquitto를 빌드하려고 하면 다음과 같은 에러가 발생할 수 있다.


In file included from /home/charsyam/mosquitto-1.3.5/lib/logging_mosq.c:34:0:
/home/charsyam/mosquitto-1.3.5/lib/mosquitto_internal.h:51:20: fatal error: ares.h: No such file or directory

이유는 ares.h가 없다는 것인데, DNS Lookup의 SRV랑 연관이 있다는데 이것이 무엇인지는 잘 모른다는 ㅋㅋㅋ

빌드를 손쉽게 하는 방법은 두가지가 있다.

1.  config.mk 에서 WITH_SRV를 찾아서 yes -> no로 바꾼다. 그리고 make

2. ares.h를 채워주면 된다. 우분투라면 apt-get install libc-ares2 libc-ares-dev로 간단하게 설치 가능 그리고 빌드하면 됨.

[입개발] Redis Scan은 어떻게 동작할까? PART #3(결)

PART #1, PART #2를 보면 결국 Redis Scan에서의 Cursor는 bucket 을 검색해야할 다음 index 값이라고 볼 수 있습니다. 그런데 실제로 실행시켜보면, 0, 1, 2 이렇게 증가하지 않고…

그 이유중에 하나는… 실제 Cursor 값이 다음 index의 reverse 값을 취하고 있기 때문입니다. 이걸 보기 전에 먼저 다시 한번 Scan의 핵심 함수인 distScan을 살펴보도록 하겠습니다.(젤 뒤만 보면 됩니다.)

unsigned long dictScan(dict *d,
                       unsigned long v,
                       dictScanFunction *fn,
                       void *privdata)
{
    dictht *t0, *t1;
    const dictEntry *de;
    unsigned long m0, m1;

    if (dictSize(d) == 0) return 0;

    if (!dictIsRehashing(d)) {
        t0 = &(d->ht[0]);
        m0 = t0->sizemask;

        /* Emit entries at cursor */
        de = t0->table[v & m0];
        while (de) {
            fn(privdata, de);
            de = de->next;
        }

    } else {
      ......    
    }

    /* Set unmasked bits so incrementing the reversed cursor
     * operates on the masked bits of the smaller table */
    v |= ~m0;

    /* Increment the reverse cursor */
    v = rev(v);
    v++;
    v = rev(v);

    return v;
}

한 이터레이션이 끝나고 나면 m0 의 bitwise NOT을 or 하고 reverse를 취한 다음 1을 더하고 다시 reverse를 취합니다. 일단 bucket이 4개만 있다고 가정하고, rehashing은 빼고 생각해보도록 합니다. 먼저 여기서 reverse는 비트를 쭈욱 세워놓고, 그걸 거꾸로 뒤집는 것입니다.
그래서 0의 rev(0) 은 그대로 0이고, rev(1)은 8000000000000000(16진수), rev(2)는 4000000000000000(16진수) 가 됩니다.(아 이걸 출력을 64bit 라 64bit hex로 찍어야 하는데 32bit로 찍었다가… 잘못된 이해를 ㅋㅋㅋ)

처음에는 v(cursor) 가 0입니다. scan이 끝나고 (0 |= ~3) = -4, 그 뒤에 rev(-4)는 3fffffffffffffff(16진수) 가 됩니다. 여기에 1을 더하면 4000000000000000 여기서 다시 rev(4000000000000000)가 되면 2가 나오게 됩니다.

/* Function to reverse bits. Algorithm from:
 * http://graphics.stanford.edu/~seander/bithacks.html#ReverseParallel */
static unsigned long rev(unsigned long v) {
    unsigned long s = 8 * sizeof(v); // bit size; must be power of 2
    unsigned long mask = ~0;
    while ((s >>= 1) > 0) {
        mask ^= (mask << s);
        v = ((v >> s) & mask) | ((v << s) & ~mask);
    }
    return v;
}

그런데 왜 reverse를 취하는 것일까요? 이것은 실제 적으로 1씩 증가하는 형태라면… cursor가 언제 끝나는지 알려주기가 애매해서 입니다. 즉 끝났다는 값을 다시 줘야 하는데, 그것보다는 0으로 시작해서 다시 0으로 끝날 수 있도록 reverse 형태를 취하는 것이죠.

PART #1, PART #2, PART #3 의 이유로 해서 SCAN은 다음과 같은 제약 사항을 가집니다.
1. count 값을 줄 수 있지만, 딱 그 개수를 보장하지 않는다.
2. 이미 scan 이 지나간 인덱스에 있는 index 에 나중에 추가된 아이템은 iteration 중에 데이터가 나오지 못한다.(Cursor가 이미 지나갔으므로…)
3. 해당 코드의 설명을 보면… 몇몇 데이터가 중복 될 수 있다는데… 이 부분은 저도 잘 이해가 안가는… 코멘트에 보면… hash table이 확장될때, 줄어들 때, rehashing 할 때 다시 스캔하지 않는다고 되어있는데… 이 부분은 잘 모르겠네요. ㅎㅎㅎ

그럼 Redis Scan 에 대한 부분은 마치도록 하겠습니다.

[입개발] Redis Scan은 어떻게 동작할까? PART #2

지난 Part #1 에서는 기본적인 Redis 의 Scan 동작과 테이블에 대해서 알아보았습니다. 이번에는 Redis Scan의 동작을 더 분석하기 위해서 기본적으로 Redis Hash Table의 Rehash 과 그 상황에서 Scan이 어떻게 동작하는지 알아보도록 하겠습니다.

먼저, Redis Hash Table의 Rehashing에 대해서 알아보도록 하겠습니다. 전 편에서도 간단하게 언급했지만 Redis Hash Table은 보통 Dynamic Bucket 에 충돌은 list 로 처리하는 방식입니다.

redis_hash_1

처음에는 4개의 Bucket으로 진행하면 Hash 값에 bitmask를 씌워서 Hash Table 내의 index를 결정합니다. 그런데, 이대로 계속 데이터가 증가하면, 당연히 충돌이 많고, List가 길어지므로, 탐색 시간이 오래걸리게 되어서 문제가 발생합니다. Redis는 이를 해결하기 위해서 hash table의 사이즈를 2배로 늘리는 정책을 취합니다.

redis_hash_expand

2배로 테이블이 늘어나면서, bitmask는 하나더 사용하도록 됩니다. 이렇게 테이블이 확장되면 Rehash를 하게 됩니다. 그래야만 검색시에 제대로 찾을 수 있기 때문입니다. 먼저 Table을 확장할 때 사용하는 것이 _dictExpandIfNeeded 합수입니다. dictIsRehashing는 이미 Rehash 중인지를 알려주는 함수이므로, Rehashing 중이면 이미 테이블이 확장된 상태이므로 그냥 DICT_OK를 리턴합니다.

먼저 hash table에서 hash table의 사용 정도가 dict_force_resize_ratio 값 보다 높으면 2배로 확장하게 됩니다.

/* Expand the hash table if needed */
static int _dictExpandIfNeeded(dict *d)
{
    /* Incremental rehashing already in progress. Return. */
    if (dictIsRehashing(d)) return DICT_OK;

    /* If the hash table is empty expand it to the initial size. */
    if (d->ht[0].size == 0) return dictExpand(d, DICT_HT_INITIAL_SIZE);

    /* If we reached the 1:1 ratio, and we are allowed to resize the hash
     * table (global setting) or we should avoid it but the ratio between
     * elements/buckets is over the "safe" threshold, we resize doubling
     * the number of buckets. */
    if (d->ht[0].used >= d->ht[0].size &&
        (dict_can_resize ||
         d->ht[0].used/d->ht[0].size > dict_force_resize_ratio))
    {
        return dictExpand(d, d->ht[0].used*2);
    }
    return DICT_OK;
}

실제로 _dictExpandIfNeeded 는 _dictKeyIndex 함수에서 호출하게 됩니다. 이렇게 테이블이 확장되면 Rehash를 해야 합니다. Rehash 라는 것은 테이블의 Bucket 크기가 커졌고 bitmask 가 달라졌으니… mask 0011이 전부 3번째 index였다면 이중에서 111은 7번째로, 011은 3번째로 옮기는 것입니다. 여기서 Redis의 특징이 하나 있습니다. 한꺼번에 모든 테이블을 Rehashing 해야 하면 당연히 시간이 많이 걸립니다. O(n)의 시간이 필요합니다. 그래서 Redis는 rehash flag와 rehashidx라는 변수를 이용해서, hash table에서 하나씩 Rehash하게 됩니다. 즉, 확장된 크기가 8이라면 이전 크기 총 4번의 Rehash 스텝을 통해서 Rehashing이 일어나게 됩니다. (이로 인해서 뒤에서 설명하는 특별한 현상이 생깁니다.)

그리고 현재 rehashing 중인것을 체크하는 함수가 dictIsRehashing 함수입니다. rehashidx 가 -1이 아니면 Rehashing 중인 상태입니다.

#define dictIsRehashing(d) ((d)->rehashidx != -1)

그리고 위의 _dictExpandIfNeeded 에서 호출하는 실제 hash table의 크기를 증가시키는 dictExpand 함수에서 rehashidx를 0으로 설정합니다.

/* Expand or create the hash table */
int dictExpand(dict *d, unsigned long size)
{
    dictht n; /* the new hash table */
    unsigned long realsize = _dictNextPower(size);

    /* the size is invalid if it is smaller than the number of
     * elements already inside the hash table */
    if (dictIsRehashing(d) || d->ht[0].used > size)
        return DICT_ERR;

    /* Allocate the new hash table and initialize all pointers to NULL */
    n.size = realsize;
    n.sizemask = realsize-1;
    n.table = zcalloc(realsize*sizeof(dictEntry*));
    n.used = 0;

    /* Is this the first initialization? If so it's not really a rehashing
     * we just set the first hash table so that it can accept keys. */
    if (d->ht[0].table == NULL) {
        d->ht[0] = n;
        return DICT_OK;
    }

    /* Prepare a second hash table for incremental rehashing */
    d->ht[1] = n;
    d->rehashidx = 0;
    return DICT_OK;
}

위의 함수를 잘 살펴보면 dict 구조체 안의 ht[1] = n으로 할당하는 코드가 있습니다. 이 얘기는 hash table이 두 개라는 것입니다. 먼저 dict 구조체를 살펴보면 다음과 같습니다.

typedef struct dictht {
    dictEntry **table;
    unsigned long size;
    unsigned long sizemask;
    unsigned long used;
} dictht;

typedef struct dict {
    dictType *type;
    void *privdata;
    dictht ht[2];
    long rehashidx; /* rehashing not in progress if rehashidx == -1 */
    int iterators; /* number of iterators currently running */
} dict;

실제로, redis의 rehashing 중에는 Hash Table이 두개가 존재합니다. 이것은 앞에 설명했듯이… 한번에 rehash step이 끝나지 않고, 매번 하나의 bucket 별로 rehashing을 하기 때문입니다. 즉 hash table의 확장이 일어나면 다음과 같이 두 개의 hash table 이 생깁니다.

redis_hash_expand_1

그리고 한스텝이 자나갈 때 마다 하나의 Bucket 단위로 해싱이 됩니다. 즉 첫번째 rehash step에서는 다음과 같이 ht[0]에 있던 데이터들이 ht[1]으로 나뉘어서 들어가게 됩니다.

redis_hash_rehash_1

두 번째, 세 번째, 네 번째 rehash 스텝이 끝나면 완료되게 됩니다.

redis_hash_rehash_2

그럼 의문이 생깁니다. Rehashing 중에 추가 되는 데이터는? 또는 삭제나 업데이트는? 추가 되는 데이터는 이 때는 무조건 ht[1]으로 들어가게 됩니다.(또 해싱 안해도 되게…), 두 번째로, 검색이나, 업데이트는, 이 때 ht[0], ht[1]을 모두 탐색하게 됩니다.(어쩔 수 없겠죠?)

dictRehash 함수에서 이 rehash step을 처리하게 됩니다. dictRehash 함수의 파라매터 n은 이 스텝을 몇번이나 할 것인가 이고, 실제로 수행할 hash table의 index는 함수중에서 ht[0]의 table이 NULL인 부분을 스킵하면서 찾게 됩니다. 그리고 ht[0]의 used 값이 0이면 rehash가 모두 끝난것이므로 ht[1]을 ht[0]로 변경하고 rehashidx를 다시 -1로 셋팅하면서 종료하게 됩니다.

/* Performs N steps of incremental rehashing. Returns 1 if there are still
 * keys to move from the old to the new hash table, otherwise 0 is returned.
 * Note that a rehashing step consists in moving a bucket (that may have more
 * than one key as we use chaining) from the old to the new hash table. */
int dictRehash(dict *d, int n) {
    if (!dictIsRehashing(d)) return 0;

    while(n--) {
        dictEntry *de, *nextde;

        /* Check if we already rehashed the whole table... */
        if (d->ht[0].used == 0) {
            zfree(d->ht[0].table);
            d->ht[0] = d->ht[1];
            _dictReset(&d->ht[1]);
            d->rehashidx = -1;
            return 0;
        }

        /* Note that rehashidx can't overflow as we are sure there are more
         * elements because ht[0].used != 0 */
        assert(d->ht[0].size > (unsigned long)d->rehashidx);
        while(d->ht[0].table[d->rehashidx] == NULL) d->rehashidx++;
        de = d->ht[0].table[d->rehashidx];
        /* Move all the keys in this bucket from the old to the new hash HT */
        while(de) {
            unsigned int h;

            nextde = de->next;
            /* Get the index in the new hash table */
            h = dictHashKey(d, de->key) & d->ht[1].sizemask;
            de->next = d->ht[1].table[h];
            d->ht[1].table[h] = de;
            d->ht[0].used--;
            d->ht[1].used++;
            de = nextde;
        }
        d->ht[0].table[d->rehashidx] = NULL;
        d->rehashidx++;
    }
    return 1;
}

이제 다시 scan으로 돌아오면… Rehashing 중의 dictScan 함수는 다음과 같습니다.

    } else {
        t0 = &d->ht[0];
        t1 = &d->ht[1];

        /* Make sure t0 is the smaller and t1 is the bigger table */
        if (t0->size > t1->size) {
            t0 = &d->ht[1];
            t1 = &d->ht[0];
        }

        m0 = t0->sizemask;
        m1 = t1->sizemask;

        /* Emit entries at cursor */
        de = t0->table[v & m0];
        while (de) {
            fn(privdata, de);
            de = de->next;
        }

        /* Iterate over indices in larger table that are the expansion
         * of the index pointed to by the cursor in the smaller table */
        do {
            /* Emit entries at cursor */
            de = t1->table[v & m1];
            while (de) {
                fn(privdata, de);
                de = de->next;
            }

            /* Increment bits not covered by the smaller mask */
            v = (((v | m0) + 1) & ~m0) | (v & m0);

            /* Continue while bits covered by mask difference is non-zero */
        } while (v & (m0 ^ m1));
    }

실제로 이미 Rehashing이 된 bucket의 경우는 ht[0] 작은 hash table에는 이미 index의 값이 NULL이므로 실제로 돌지 않지만, 아직 rehash되지 않은 bucket의 경우는 ht[0] 와 ht[1] 의 두 군데, 즉 총 세 군데에 데이터가 존재할 수 있습니다. 그래서 먼저 ht[0]의 bucket을 돌고 나서, ht[1]을 찾게 됩니다. 여기서 당연히 ht[1]에서는 두 군데를 검색해야 하므로 두 번 돌게 됩니다.

v = (((v | m0) + 1) & ~m0) | (v & m0);

즉 위의 식은 만약 v가 0이고 m0 = 3, m1 = 7이라고 하면… (((0 | 3) + 1) & ~3) | (0 & 3) 이 되게 됩니다. ~3은 Bitwise NOT 3이 되므로 -4 가 나오고, (4 & -4) | 0 이므로 결론은 4 & -4 입니다. 3은 00000011, bitwise NOT하면 11111100 이 되므로, 즉 다시 풀면, 00000100 & 11111100 해서 00000100 즉 4가 나오게됩니다. 처음에는 index 0, 두번째는 index 4 가 되는 거죠. 그래서 첫 루프를 돌게 됩니다. 다시 4 & (m0 ^ m1) == 4 이므로 …

이제 두 번째 루프에서 다시 (((4 | 3) + 1) & -4) | (4 & 3) 이므로… 4 | 3 = 7, 4 & 3 = 0 이므로… 다시 한번 정리하면 ((7+1) & -4) | 0 이므로 결론은 8 & -4 = 4 가 되므로 00001000 & 111111100 이 되므로 v 는 이번에는 00001000 즉 8이 됩니다. 즉 한번 돌 때 마다, ht[0]의 size 만큼 증가하게 됩니다.(다들 한방에 이해하실 텐데… 이걸 설명한다고…) 그래서 그 다음번에는 8 & 4 가 되므로 루프가 끝나게 됩니다. 즉, 0, 4 이렇게 ht[1]에서 두 번 읽어야 하니, 두 번 읽는 코드를 만들어둔거죠.

이제 다음편에는 cursor 가 어떻게 만들어지는가에 대해서 간단하게 설명하도록 하겠습니다. (다음편은 짧을듯…)

[입 개발] Redis Scan은 어떻게 동작할까? PART #1

처음부터 꾸준히 입만 놀리는 입개발(or 혀로그래머) CharSyam입니다. Redis의 기능중에 사람들이 쓰면 안되지만, 그 단맛에 끌려 어쩔 수 없이 치게 되는 명령이 KEYS 입니다. KEYS를 쓰는 순간, Redis는 이 명령을 처리하기 위해서 멈춰버립니다. 특히 트래픽이 많은 서버는… 이 KEYS 명령 하나에 많은 장애를 내게 됩니다. 그런데… 어느 순간… Redis에 Scan이라는 명령이 생겼습니다. KEYS의 단점을 없애면서도, 느리지 않은… 그렇다면, Redis에서는 어떻게 Scan 이 동작하게 될까요?

Scan의 내부 동작을 알기 전에… 먼저 Redis가 어떻게 데이터를 저장하는지 부터 다시 한번 집고 넘어가야 합니다. Redis 의 가장 기초적인 자료구조는 KV 즉 Key/Value 형태를 저장하는 것입니다.(String 타입이라고도 합니다.) 이를 위해 Redis는 Bucket을 활용한 Chained Linked List 구조를 사용합니다. 최초에는 4개의 Bucket에서 사용하여… 같은 Bucket에 들어가는 Key는 링크드 리스트 형태로 저장하는 거죠. 즉 다음 그림과 같습니다.

redis_hash_1

이 Chained Linked List에는 다음과 같은 약점이 있습니다. 즉 한 Bucket 안에 데이터가 많아지면 결국 탐색 속도가 느려집니다. 이를 위해서 Redis는 특정 사이즈가 넘을때마다 Bucket을 두 배로 확장하고, Key들을 rehash 하게 됩니다. 먼저 이 때 Key의 Hash로 사용하는 해시함수는 다음과 같습니다. MurmurHash2를 사용합니다.

/* MurmurHash2, by Austin Appleby
 * Note - This code makes a few assumptions about how your machine behaves -
 * 1. We can read a 4-byte value from any address without crashing
 * 2. sizeof(int) == 4
 *
 * And it has a few limitations -
 *
 * 1. It will not work incrementally.
 * 2. It will not produce the same results on little-endian and big-endian
 *    machines.
 */
unsigned int dictGenHashFunction(const void *key, int len) {
    /* 'm' and 'r' are mixing constants generated offline.
     They're not really 'magic', they just happen to work well.  */
    uint32_t seed = dict_hash_function_seed;
    const uint32_t m = 0x5bd1e995;
    const int r = 24;

    /* Initialize the hash to a 'random' value */
    uint32_t h = seed ^ len;

    /* Mix 4 bytes at a time into the hash */
    const unsigned char *data = (const unsigned char *)key;

    while(len >= 4) {
        uint32_t k = *(uint32_t*)data;

        k *= m;
        k ^= k >> r;
        k *= m;

        h *= m;
        h ^= k;

        data += 4;
        len -= 4;
    }

    /* Handle the last few bytes of the input array  */
    switch(len) {
    case 3: h ^= data[2] << 16;
    case 2: h ^= data[1] << 8;
    case 1: h ^= data[0]; h *= m;
    };

    /* Do a few final mixes of the hash to ensure the last few
     * bytes are well-incorporated. */
    h ^= h >> 13;
    h *= m;
    h ^= h >> 15;

    return (unsigned int)h;
}

그리고 hash 값이 들어가야 할 hash table 내의 index를 결정하는 방법은 다음과 같습니다.

/* Returns the index of a free slot that can be populated with
 * a hash entry for the given 'key'.
 * If the key already exists, -1 is returned.
 *
 * Note that if we are in the process of rehashing the hash table, the
 * index is always returned in the context of the second (new) hash table. */
static int _dictKeyIndex(dict *d, const void *key)
{
    ......
    h = dictHashKey(d, key);
    for (table = 0; table <= 1; table++) {
        idx = h & d->ht[table].sizemask;
        ......
    }
    return idx;
}

table에는 Key를 찾기위해 비트 연산을 하기 위한 sizemask가 들어가 있습니다. 초기에는 table의 bucket이 4개 이므로 sizemask는
이진수로 11 즉 3의 값이 셋팅됩니다. 즉 해시된 결과 & 11의 연산결과로 들어가야 하는 Bucket이 결정되게 됩니다.

여기서 Key 가 많아지면 Redis는 Table의 사이즈를 2배로 늘리게 됩니다. 그러면 당연히 sizemask도 커지게 됩니다. Table size가 8이면 sizemask는 7이 됩니다.

먼저 간단하게 말하자면, Scan의 원리는 이 Bucket을 한 턴에 하나씩 순회하는 것입니다. 그래서 아래 그림과 같이 처음에는 Bucket Index 0를 읽고 데이터를 던져주는 것입니다.

redis_scan_0

        t0 = &(d->ht[0]);
        m0 = t0->sizemask;

        /* Emit entries at cursor */
        de = t0->table[v & m0];
        while (de) {
            fn(privdata, de);
            de = de->next;
        }

다음 편에서는 실제 사용되는 cursor 가 어떤 식으로 만들어지는 지, 그 외의 예외 케이스는 어떤 것이 있는지… 그리고 scan 을 사용할 때 주의사항에는 어떤 것이 있는지 알아보도록 하겠습니다.(이거 적고, 안 쓸지도 ㅎㅎㅎ)