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

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

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

Для создания персистентной очереди очень удобно пользоваться ее реализацией на стеках, поскольку стеки легко сделать персистентными. Реализация на двух стеках не подходит для этого, так как в худшем сучае требует [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>