/D/werner/TxMemory/sources/txmem.cpp

Go to the documentation of this file.

/****
 Persistent Transactional Memory v0.3
 Copyright (C) 2010 Werner Van Belle
 http://active.yellowcouch.org/

 This program is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published by
 the Free Software Foundation; either version 2 of the License, or
 (at your option) any later version.

 This program is distributed in the hope that it will be useful,
 but WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 GNU General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with this program; if not, write to the Free Software
 Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
****/
#ifndef __loaded__txmem_cpp__
#define __loaded__txmem_cpp__
using namespace std;
#line 1 "txmem.c++"
#include <time.h>
#include <errno.h>
#include <unistd.h>
#include <sys/types.h>
#include <assert.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include "pagemap.h"
#include <vector>
#include "txmem.h"
using namespace std;
//#define DEBUG
#define RF 10
#define RANDOM_BREAKS
#define RIGID_FLAGS

TxMemory::TxMemory(const char* filename, unsigned long size)
{
  pagesize=getpagesize();
  assert(pagesize>0);
  pagecount=1+(size-1)/pagesize;
#ifdef DEBUG
  printf("Requires %d pages\n",pagecount);
#endif
  memory_size=pagecount*pagesize;
  
  // open the journal 
  char T[1000];
  sprintf(T,"%s.journal",filename);
  journalfn=strdup(T);
  sprintf(T,"%s.journal.tmp",filename);
  journaltmp=strdup(T);
  sprintf(T,"%s.memory",filename);
  char* memoryfn=strdup(T);

  int journal=open(journalfn,O_RDWR);
  if (journal==-1 && errno==ENOENT)
    journal=0;
  assert(journal>=0);
  
  // open the memory file
  int err;
  fd=open(memoryfn, O_RDWR);
  if (fd<=0)
    {
      fd=open(memoryfn, O_CREAT|O_RDWR,S_IRUSR|S_IWUSR);
#ifdef DEBUG
      printf("Created memory file %s\n",memoryfn);
#endif
      assert(fd);
      // set to appropriate length
      err=ftruncate(fd,memory_size);
      assert(!err);
      ftruncate(journal,0);
    }
  assert(fd);
  
  // restore journal if it is present
  if (journal)
    {
      long long journal_size=lseek(journal,0,SEEK_END);
      assert(journal_size>sizeof(int));
      int journalpagesize;
      lseek(journal,0,SEEK_SET);
      int R=read(journal,&journalpagesize,sizeof(int));
      assert(R==sizeof(int));
      printf("Reading memory journal");
      fflush(stdout);
      int blocksize=journalpagesize+sizeof(int);
      assert((journal_size-sizeof(int))%blocksize==0);
      int entries=(journal_size-sizeof(int))/blocksize;
      for(int i = 0 ; i < entries; i++)
        {
          int journalpos=sizeof(int)+i*blocksize;
          int pos=lseek(journal,journalpos,SEEK_SET);
          assert(pos==journalpos);
          int pageno;
          char buffer[journalpagesize];
          R=read(journal,&pageno,sizeof(int));
          assert(R==sizeof(int));
          R=read(journal,buffer,journalpagesize);
          assert(R==journalpagesize);
          pos=lseek(fd,pageno*journalpagesize,SEEK_SET);
          assert(pos==pageno*journalpagesize);
          printf(" %d",pageno);
          int W=write(fd,buffer,journalpagesize);
#ifdef RANDOM_BREAKS
          if (random()%(entries*RF)==0)
            exit(13);
#endif
          assert(W==journalpagesize);
        }
      err=fdatasync(fd);
      assert(!err);
      err=close(journal);
      assert(!err);
#ifdef RANDOM_BREAKS
      if (random()%(4*RF)==0)
        exit(14);
#endif
      err=remove(journalfn);
      assert(!err);
      printf(" [done]\n");
    }
  
  // map file in memory as a private map, so that 
  M=NULL;
  reopen_map();
  
  // prepare tracking of copy-on-write pages
  mapfd=open("/proc/self/pagemap",O_RDONLY);
  assert(mapfd>0);
  pageflags=open("/proc/kpageflags",O_RDONLY);
  assert(pageflags>0);    
}

void TxMemory::reopen_map()
{
  if (M)
    munmap(M,memory_size);
  char* bef=M;
  M=(char*)mmap(M, memory_size, PROT_READ|PROT_WRITE, MAP_PRIVATE,fd,0);
  assert(M!=(char*)-1);
  assert(M);
  if (bef) assert(bef==M);
};

// commit all local changes to disk
unsigned long long TxMemory::commit()
{
  unsigned long long target=((unsigned long)(void*)M);
  //  printf("Seeking to page %llx pagesize: %lld\n",target,pagesize);
  target/=pagesize;
  //  printf("Seeking to page %lx\n",target);
  int err=lseek64(mapfd, target*8, SEEK_SET);
  assert(err==target*8);
  assert(sizeof(long long)==8);
  unsigned long long page2pfn[pagecount];
  err=read(mapfd,page2pfn,sizeof(long long)*pagecount);
  if (err<0)
    perror("Reading pagemap");
  if(err!=pagecount*8)
    printf("Could only read %d bytes\n",err);
  
  vector<int> changed_pages;
  for(int i = 0 ; i < pagecount; i++)
    {
      unsigned long long v2a=page2pfn[i];
      if (v2a==0) continue; // this page is not allocated
      int this_page_size=(PM_PSHIFT_MASK&v2a)>>PM_PSHIFT_OFFSET;
      assert(1<<this_page_size==pagesize);
      v2a&=~PM_PSHIFT_MASK;
      //      printf("Page: %d, size %llx\n",i,);
      if(v2a&0x8000000000000000LL) // the page is present in memory
        {
          unsigned long long pfn=PM_PFRAME(v2a);
          // the page is present. Now we would like to know at what address
          err=lseek64(pageflags,pfn*8,SEEK_SET);
          assert(err==pfn*8);
          unsigned long long pf;
          err=read(pageflags,&pf,8);
          assert(pf&(1LL<<KPF_MMAP));
          pf&=~(1LL<<KPF_MMAP);
          pf&=~(1LL<<KPF_LRU);
          pf&=~(1LL<<KPF_ACTIVE);
          if ((pf&(1<<KPF_SWAPBACKED)) || (pf&(1<<KPF_SWAPCACHE)))
            {
              changed_pages.push_back(i);
#ifdef RIGID_FLAGS
              if (pf!=0x5008)
                {
                  printf("Unknown page flags 1 %llx\n",pf);
                  fflush(stdout);
                  exit(10);
                }
#endif
            }
#ifdef RIGID_FLAGS
          else
            {
              // assume it is a private mapping
              // and then we don't care about that anymore
              pf&=~(1LL<<KPF_PRIVATE);      
              
              // the page was referenced (that would be 'read' ?)
              // don't care about that
              pf&=~(1LL<<KPF_REFERENCED); 
              
              // the page should be up to date in all cases.
              assert(pf&(1LL<<KPF_UPTODATE));
              pf&=~(1LL<<KPF_UPTODATE); 
              pf&=~(1LL<<KPF_MAPPEDTODISK); // if it is mapped to disk, all the better !
              if (pf!=0)
                {
                  printf("Unknown page flags 2 %llx\n",pf);
                  fflush(stdout);
                  exit(10);
                }
            }
#endif
        }
      else if (v2a==0) continue;
      //      else if (v2a&0x4000000000000000LL) // the page is in swapspace, meaning that we very likely wrote to it
      //        changed_pages.push_back(i);
      else // the page is not allocated at all
        printf("Page: %d, flag %llx\n",i,page2pfn[i]);
    }
  if (changed_pages.size()==0) return 0;
  // save pages to journal
  printf("Writing memory journal"); fflush(stdout);
  int journal=open(journaltmp,O_CREAT|O_RDWR|O_TRUNC,S_IRUSR|S_IWUSR);
  assert(journal>0);
  int pos=lseek(journal,0,SEEK_END);
  assert(pos==0);
  int ps=pagesize;
  int w=write(journal,&ps,sizeof(int));
  assert(w==sizeof(int));
  for(int j = 0 ; j < changed_pages.size(); j++)
    {
      int pageno=changed_pages[j];
      int written=write(journal,&pageno,sizeof(int));
      assert(written==sizeof(int));
      written=write(journal,M+(pageno*pagesize),pagesize);
      assert(written==pagesize);
    }
#ifdef RANDOM_BREAKS
  if (random()%(4*RF)==0)
    exit(10);
#endif
  err=fdatasync(journal);
  assert(!err);
  err=close(journal);
  assert(!err);
  err=rename(journaltmp,journalfn);
  assert(!err);
  printf(" [done]\n");
  // now start dumping the pages that changed
  printf("Committing pages:");
  for(int j = 0 ; j < changed_pages.size(); j++)
    {
      int pageno=changed_pages[j];
      printf(" %d",pageno);
      unsigned long long pos=lseek(fd,pageno*pagesize,SEEK_SET);
      assert(pos==pageno*pagesize);
      // then write our page out
#ifdef RANDOM_BREAKS
      if (random()%(changed_pages.size()*RF*2)==0)
        exit(11);
#endif
      int written=write(fd,M+pageno*pagesize,pagesize);
      assert(written==pagesize);
    }
  printf(" [done]\n");
  err=fdatasync(fd);
  assert(!err);
  
  // delete journal
#ifdef RANDOM_BREAKS
  if (random()%(RF*4)==0)
    exit(12);
#endif
  err=remove(journalfn);
  assert(!err);
  
  reopen_map();
  return changed_pages.size()*pagesize;
}

void TxMemory::abort()
{
  reopen_map();
}

// TODO
// - growing of the file
// - growing of the memory
// - checking whether the so called 'non-dirty' pages are indeed still the same
// - writing test program to access the data
#endif // __loaded__txmem_cpp__

---