Персистентная очередь

Персистентная очередь — это очередь, реализующая персистентность, то есть позволяющая получить доступ ко всем своим предыдущим версиям. Как будет показано далее, можно реализовать функциональную персистентность, то есть каждая ячейка памяти в такой структуре будет инициализирована один раз и в дальнейшем не изменится.

Основная идея

Для создания персистентной очереди очень удобно пользоваться ее реализацией на стеках, поскольку стеки легко сделать персистентными, причем в этом случае мы добьемся функциональной персистетнтности. Реализация на двух стеках не подходит для этого, так как в худшем сучае требует [math]O(n)[/math] времени, а значит и [math]O(n)[/math] памяти в случае персистентности на операцию. Покажем сначала как создать очередь в реальном времени с [math]O(1)[/math] времени на операцию, а затем превратим ее в персистентную.

Реализация очереди на шести стеках

Одним из минусов реализации на двух стеках является то, что в худшем случае мы тратим [math]O(n)[/math] времени на операцию. Если распределить время, необходимое для перемещения элементов из одного стека в другой, по операциям, мы получим очередь без худших случаев с [math]O(1)[/math] истинного времени на операцию.

Сначала будем действовать аналогично случаю с двумя стеками. Пусть у нас есть стек [math]L[/math] для операций [math]push[/math] и стек [math]R[/math] для операций [math]pop[/math]. К моменту опустошения стека [math]R[/math] нам нужно успеть получить стек [math]R'[/math], содержащий текущие элементы стека [math]L[/math] в правильном для извлечения порядке. Перекопирование (recopy mode) начнется, когда появится опасность того, что мы не сможем за оставшиеся [math]R.size[/math] операций [math]pop[/math] со стеком [math]R[/math] перекопировать стек [math]L[/math] в новый стек [math]R'[/math]. Очевидно, это ситуация [math]L.size\gt R.size[/math], пусть такое состояние отражает специальная переменная логического типа [math]recopy[/math].

Понятно, что во время перекопирования могут поступить операции [math]push[/math], а стек [math]L[/math] в это время потеряет свою структуру, сложить элементы туда мы уже не сможем, значит нужно завести еще один стек [math]L'[/math], в который мы и будем складывать новые элементы. После окончания перекопирования мы поменяем ролями [math]L,L'[/math] и [math]R,R'[/math], на первый взгляд, все станет хорошо.

Однако, если реализовать этот алгоритм, мы получим неприятную вещь: старый стек [math]R[/math] может и не опустошиться за это время, то есть мы получили два стека с выходными данными, а значит, возможен случай (например, если все поступающие операции — [math]push[/math]), когда при следующем перекопировании у нас не будет свободного стека для копировании туда элементов [math]L[/math]. Для преодоления этой проблемы мы принудительно будем извлекать все элементы из стека [math]R[/math] во вспомогательный стек [math]T[/math], затем копировать элементы из стека [math]L[/math] в [math]R[/math], а затем обратно копировать элементы из стека [math]T[/math] в [math]R[/math]. Легко показать, что приведенный алгоритм как раз получает на выходе в [math]R[/math] все элементы стеков [math]L,R[/math] в правильном порядке.

Но этого еще недостаточно. Если мы принудительно извлекаем элементы из стека [math]R[/math], появляются следующие проблемы:

1. Что вернуть при операции [math]pop[/math]? Для этого заведем себе стек [math]Rc[/math] — копию стека [math]R[/math], из которого мы и будем извлекать требуемые элементы.
2. Как поддерживать корректность такой копии? Поскольку этот стек нужен только для перекопирования, а во время него он занят, нужна запасная копия [math]Rc'[/math] для копирования всех элементов, которые мы копируем в [math]R[/math], а по окончании перекопирования поменяем ролями стеки [math]Rc, Rc'[/math], как мы делали со стеками [math]L, L'[/math].
3. Как учесть, что во время перекопирования часть элементов была извлечена из [math]Rc[/math]? Для этого заведем специальную переменную [math]toCopy[/math], которая показывает, сколько корректных элементов находится в стеке [math]T[/math], и уменьшается при каждом извлечении из [math]T[/math] или операции [math]pop[/math]. К счастью, все некорректные элементы будут нарастать со дна стека, так что мы никогда не извлечем некорректный элемент, если [math]toCopy\gt 0[/math]. Если во время операции [math]pop[/math] у нас [math]toCopy = 0[/math], это означает, что теперь в стеке [math]R[/math] находится весь правый кусок очереди, так что нам придется извлечь элемент из него.

Теперь может возникнуть проблема с непустым [math]Rc[/math] после завершения перекопирования. Покажем, что мы всегда успеем его опустошить, если будем использовать дополнительное извлечение из него при каждой операции в обычном режиме, для этого полностью проанализируем алгоритм.

Пусть на начало перекопирования в стеке [math]R[/math] содержится [math]n[/math] элементов, тогда в стеке [math]L[/math] находится [math]n+1[/math] элементов. Мы корректно можем обработать любое количество операций [math]push[/math], а также [math]n[/math] операций [math]pop[/math]. Заметим, что операция [math]empty[/math] во время перекопирования всегда возвращает [math]false[/math], так как мы не можем извлекать элементы из стека [math]L[/math], который не пустой. Таким образом вместе с операцией, активирующей перекопирование, мы гарантированно можем корректно обработать [math]n + 1[/math] операцию.

Посмотрим на дополнительные действия, которые нам предстоят:

1. Переместить содержимое [math]R[/math] в [math]T[/math], [math]n[/math] действий.
2. Переместить содержимое [math]L[/math] в стеки [math]R, Rc'[/math], [math]n + 1[/math] действий.
3. Переместить первые [math]toCopy[/math] элементов из [math]T[/math] в [math]R, Rc'[/math], остальные выкинуть, [math]n[/math] действий.
4. Поменять ролями стеки [math]Rc, Rc'[/math], [math]L, L'[/math], [math]2[/math] действия.

Таким образом, получили [math]3 \cdot n + 3[/math] дополнительных действия за [math]n + 1[/math] операций, или [math]3=O(1)[/math] дополнительных действий на операцию в режиме перекопирования, что и требовалось.

Теперь рассмотрим, как изменились наши стеки за весь период перекопирования. Договоримся, что операция [math]empty[/math] не меняет очередь, то есть никакие дополнительные действия не совершаются. Пусть за [math]n[/math] следующих за активацией меняющих операций ([math]push, pop[/math]) поступило [math]x[/math] операций [math]pop[/math], [math]n - x[/math] операций [math]push[/math]. Очевидно, что после перекопирования в новых стеках окажется: [math]n-x[/math] элементов в [math]L[/math], [math]2 \cdot n + 1 - x = (n - x) + (n + 1)[/math] элементов в [math]R[/math], то есть до следующего перекопирования еще [math]n+2[/math] операции. С другой стороны, стек [math]Rc[/math] содержал всего [math]n[/math] элементов, так что мы можем очистить его, просто удаляя по одному элементу при каждой операции в обычном режиме.

Итак, очередь [math]Q[/math] будет состоять из шести стеков [math]L,L',R,Rc,Rc',T[/math], а также двух внутренних переменных [math]recopy, toCopy[/math], которые нужны для корректности перекопирования + дополнительная переменная [math]copied[/math], показывающая, перемещали ли мы элементы из стека [math]L[/math] в стек [math]R[/math], чтобы не начать перемещать эти элементы в стек [math]T[/math].

Инвариант очереди (обычный режим):

1. Стек [math]L[/math] содержит левую половину очереди, порядок при извлечении обратный.
2. Стек [math]R[/math] содержит правую половину очереди, порядок при извлечении прямой.
3. [math]L.size \leqslant R.size[/math]
4. [math]R.size = 0 \equiv Q.size = 0[/math]
5. [math]Rc[/math] — копия [math]R[/math]
6. [math]Rc'.size \lt R.size - L.size[/math]
7. [math]L'.size = 0, T.size = 0[/math]

Тогда к следующему перекопированию ([math]L.size=R.size+1[/math]) мы гарантированно будем иметь пустые стеки [math]L',T,Rc'[/math], которые нам понадобятся.

Инвариант очереди (режим перекопирования):

1. [math]Rc.size = toCopy[/math]
2. Если [math]L.size = 0[/math], то:
   1. При [math]toCopy \gt 0[/math] первые [math]toCopy[/math] элементов [math]T[/math] — корректны, то есть действительно содержатся в очереди.
   2. При [math]toCopy \leqslant 0[/math] стек [math]R[/math] содержит весь правый кусок очереди в правильном порядке.

Очередь будет работать в двух режимах:

1. Обычный режим, кладем в [math]L[/math], извлекаем из [math]R[/math] и из [math]Rc, Rc'[/math] для поддержания порядка, операция [math]empty = (R.size = 0)[/math].
2. Режим перекопирования, кладем в [math]L'[/math], извлекаем из [math]Rc[/math], возможно из [math]R[/math], [math]empty=false[/math], совершаем дополнительные действия.

Также после операции в обычном режиме следует проверка на активацию перекопирования ([math]recopy = (L.size \gt R.size)[/math]), если это так, [math]toCopy=R.size, recopy=true, copied = false[/math], совершается первый набор дополнительных действий.

После операции в режиме перекопирования следует проверка на завершение перекопирования ([math]recopy=(T.size==0)[/math]), а при завершении меняются ролями стеки [math]Rc, Rc'[/math], [math]L, L'[/math].

Следующий псевдокод выполняет требуемые операции:

empty

empty()
   return !recopy and R.size == 0

push

push(x)
   if !recopy
      L.push(x)
      if Rc'.size > 0
         Rc'.pop()
      checkRecopy()
   else
      L'.push(x)
      checkNormal()

pop

pop()
   if !recopy
      tmp = R.pop()
      Rc.pop()
      if Rc'.size > 0
         Rc'.pop()
      checkRecopy()
      return tmp
   else
      tmp = Rc.pop()
      if toCopy > 0
         toCopy = toCopy - 1
      else
         R.pop()
         Rc'.pop()
      checkNormal()
      return tmp

checkRecopy

checkRecopy()    
   recopy = L.size > R.size
   if recopy
      toCopy = R.size
      copied = false
      checkNormal()

checkNormal

checkNormal()
   additionalOperations()
   // Если мы не все перекопировали, то у нас не пуст стек T
   recopy = T.size [math] \ne [/math] 0

additionalOperations

additionalOperations()
   // Нам достаточно 3 операций на вызов
   toDo = 3
   // Пытаемся перекопировать R в T
   while not copied and toDo > 0 and R.size > 0
      T.push(R.pop())
      toDo = toDo - 1
   // Пытаемся перекопировать L в R и Rc'
   while toDo > 0 and L.size > 0
      copied = true
      x = L.pop()
      R.push(x)
      Rc'.push(x)
      toDo = toDo - 1
   // Пытаемся перекопировать T в R и Rc' с учетом toCopy
   while toDo > 0 and T.size > 0
      x = T.pop()
      if toCopy > 0
         R.push(x)
         Rc'.push(x)
         toCopy = toCopy - 1
      toDo = toDo - 1
   // Если все скопировано, то меняем роли L, L' и Rc, Rc'
   if T.size == 0
      swap(L, L')
      swap(Rc, Rc')

Персистентная очередь на пяти стеках

После того, как мы получили очередь в реальном времени с [math]O(1)=6[/math] обычными стеками, ее можно легко превратить в персистентную, сделав все стеки персистентными, но на самом деле персистентность позволяет не создавать явной копии стека [math]R[/math], так что достаточно всего пяти стеков.

Вместо стеков [math]Rc, Rc'[/math] персистентная очередь хранит один стек [math]R'[/math], в который при активации перекопирования записывается последняя версия стека [math]R[/math], в дальнейшем все операции [math]pop[/math] обращаются именно к ней. Все замечания о [math]toCopy[/math] остаются в силе.

Также нам нет необходимости опустошать стек [math]R'[/math] к моменту перекопирования, так как скопировать туда новую версию [math]R[/math] мы можем за [math]O(1)[/math], а освобождение ячеек памяти бессмысленно, так как они используются в других версиях персистентной очереди.

В качестве версии очереди мы будем использовать запись [math]Q=\lt L, L', R, R', T, recopy, toCopy, copied\gt [/math], содержащую пять версий персистентных стеков и три переменных.

Пусть персистентный стек возвращает вместе с обычным результатом работы стека новую версию, то есть операция [math]S.pop[/math] возвращает [math]\lt Sn, x\gt [/math], а операция [math]S.push[/math] возвращает [math]\lt Sn\gt [/math], аналогично свою новую версию вместе с результатом операции возвращает и персистентная очередь.

Следующий псевдокод выполняет требуемые операции:

empty

empty()
   return !recopy and R.size == 0

push

push(x)
   if !recopy
      Ln = L.push(x)
      Q' = <Ln, L', R, R', T, recopy, toCopy, copied>
      return Q'.checkRecopy()
   else
      Ln' = L'.push(x)
      Q' = <L, Ln', R, R', T, recopy, toCopy, copied> 
      return Q'.checkNormal()

pop

pop()
   if !recopy
      <Rn, x> = R.pop()
      Q' = <L, L', Rn, R', T, recopy, toCopy, copied>
      return <Q'.checkRecopy(), x>
   else
      <Rn', x> = R'.pop()
      curCopy = toCopy
      Rn = R
      if toCopy > 0
         curCopy = curCopy - 1
      else
         <Rn, x> = Rn.pop() 
      Q' = <L, L', Rn, Rn', T, recopy, curCopy, copied>
      return <Q'.checkNormal(), x>

checkRecopy

checkRecopy()    
   if L.size > R.size
      Q' = <L, L', R, R', T, true, R.size, false>
      return Q'.checkNormal()
   else
      return <L, L', R, R', T, false, toCopy, copied>

checkNormal

checkNormal()
   Q' = Q.additionalOperations()
   // Если мы не все перекопировали, то у нас не пуст стек T
   return <Q'.L, Q'.L', Q'.R, Q'.R', Q'.T, Q'.T.size [math] \ne [/math] 0, Q'.toCopy, Q'.copied>

additionalOperations

additionalOperations()
   // Нам достаточно 3 операций на вызов
   toDo = 3
   // Пытаемся перекопировать R в T
   Rn = R
   Tn = T
   curCopied = copied
   while not curCopied and toDo > 0 and Rn.size > 0
      <Rn, x> = Rn.pop()
      Tn = Tn.push(x)
      toDo = toDo - 1
   Ln = L
   // Пытаемся перекопировать L в R
   while toDo > 0 and Ln.size > 0
      curCopied = true
      <Ln, x> = Ln.pop()
      Rn = Rn.push(x)
      toDo = toDo - 1
   curCopy = toCopy
   // Пытаемся перекопировать T в R с учетом toCopy
   while toDo > 0 and Tn.size > 0
      <Tn, x> = Tn.pop()
      if curCopy > 0
         Rn = Rn.push(x)
         curCopy = curCopy - 1
      toDo = toDo - 1
   Ln' = L'
   // Если все скопировано, то меняем роли L1, L2
   if T.size == 0
      swap(Ln, Ln')
   return <Ln, Ln', Rn, R', Tn, recopy, curCopy, curCopied>      

Пример

Пусть мы создали персистентную очередь. Будем изображать ее в виде пяти персистентных стеков, закрашенные вершины — текущие версии стеков, соответствующие текущему состоянию очереди; стрелка от стека [math]R'[/math] указывает на ту версию стека [math]R[/math], которая там сейчас хранится. В самих вершинах записаны соответствующие этим вершинам значения.

Сделаем операцию [math] push(1) [/math], изначально режим обычный, так что элемент пойдет в стек [math]L[/math]. Эта операция активирует режим перекопирования, в результате которого содержимое [math]L[/math] переместится в стек [math]R[/math], после чего перекопирование завершится, стеки [math]L, L'[/math] поменяются местами.

Сделаем операцию [math] push(2) [/math], у нас обычный режим, поэтому элемент пойдет в стек [math]L[/math], перекопирование не активируется.

Сделаем операцию [math] push(3) [/math], у нас обычный режим, поэтому элемент пойдет в стек [math]L[/math], активируется перекопирование, [math]R' = R[/math], за три операции мы успеваем переложить элемент стека [math]R[/math] в стек [math]T[/math], а также переложить два элемента стека [math]L[/math] в стек [math]R[/math].

Сделаем операцию [math] push(4) [/math], мы в режиме перекопирования, поэтому элемент пойдет в стек [math]L'[/math], далее мы успеваем перекопировать обратно элемент из стека [math]T[/math] в стек [math]R[/math], перекопирование завершается, стеки [math]L, L'[/math] меняются местами.

Сделаем операцию [math] push(5) [/math], у нас обычный режим, поэтому элемент пойдет в стек [math]L[/math], перекопирование не активируется.

Сделаем операцию [math] push(6) [/math], у нас обычный режим, поэтому элемент пойдет в стек [math]L[/math], перекопирование не активируется.

Сделаем операцию [math] push(7) [/math], у нас обычный режим, поэтому элемент пойдет в стек [math]L[/math], перекопирование активируется, [math]R' = R[/math], [math]toCopy=3[/math], за три операции мы успеваем переместить содержимое стека [math]R[/math] в стек [math]T[/math].

Сделаем операцию [math]pop[/math], мы находимся в режиме перекопирования, так что элемент извлекается из [math]R'[/math], [math]toCopy = 2[/math]. За три операции мы успеваем перекопировать три элемента стека [math]L[/math] в стек [math]R[/math].

Сделаем операцию [math]pop[/math], мы находимся в режиме перекопирования, так что элемент извлекается из [math]R'[/math], [math]toCopy = 1[/math]. За три операции мы успеваем перекопировать один элемент стека [math]L[/math] в стек [math]R[/math], а также извлечь два элемента стека [math]T[/math], с учетом [math]toCopy[/math] только один элемент попадет в стек [math]R[/math], [math]toCopy = 0[/math].

Сделаем операцию [math]pop[/math], мы находимся в режиме перекопирования, так что элемент извлекается из [math]R'[/math], но [math]toCopy = 0[/math], так что нам приходится извлечь еше один элемент из стека [math]R[/math]. Мы извлекаем один элемент из стека [math]T[/math], перекопирование заканчивается, стеки [math]L, L'[/math] меняются местами.

Ссылки

• Hood R., Melville R. Real Time Queue Operations in Pure LISP. — Cornell University, 1980