Robust/Transactions/TransactionalIO/src/TransactionalIO/core/ExtendedTransaction.java

   1 /*
   2  * To change this template, choose Tools | Templates
   3  * and open the template in the editor.
   4  */
   5 package TransactionalIO.core;
   6
   7 import TransactionalIO.exceptions.AbortedException;
   8 import TransactionalIO.benchmarks.benchmark;
   9 import TransactionalIO.benchmarks.customhandler;
  10 import TransactionalIO.benchmarks.customhandler;
  11 import TransactionalIO.interfaces.BlockAccessModesEnum;
  12 import TransactionalIO.interfaces.ContentionManager;
  13 import TransactionalIO.interfaces.TransactionStatu;
  14 //import dstm2.file.managers.BaseManager;
  15 import java.awt.event.ActionListener;
  16 import java.beans.EventHandler;
  17 import java.beans.PropertyChangeEvent;
  18 import java.beans.PropertyChangeListener;
  19 import java.beans.PropertyChangeSupport;
  20 import java.io.FileDescriptor;
  21 import java.io.IOException;
  22 import java.io.RandomAccessFile;
  23 import java.util.Collections;
  24 import java.util.HashMap;
  25 import java.util.Iterator;
  26 import java.util.Map;
  27 import java.util.TreeMap;
  28 import java.util.Vector;
  29 import java.util.concurrent.locks.Lock;
  30 import java.util.concurrent.locks.ReentrantLock;
  31 import java.util.concurrent.locks.ReentrantReadWriteLock;
  32 import java.util.logging.Level;
  33 import java.util.logging.Logger;
  34
  35 /**
  36  *
  37  * @author navid
  38  */
  39 public class ExtendedTransaction implements TransactionStatu {
  40
  41     private native int nativepwrite(byte buff[], long offset, int size, FileDescriptor fd);
  42
  43
  44     {
  45         System.load("/home/navid/libkooni.so");
  46     }
  47     private boolean flag = true;
  48     public TransactionStatu memorystate;
  49     private PropertyChangeSupport changes = new PropertyChangeSupport(this);
  50     public int starttime;
  51     public int endtime;
  52     public TreeMap msg = new TreeMap();
  53     public int numberofwrites;
  54     public int numberofreads;
  55
  56     public enum Status {
  57
  58         ABORTED, ACTIVE, COMMITTED
  59     };
  60     private boolean writesmerged = true;
  61     //private Vector<ReentrantLock> heldoffsetlocks;
  62     private Vector heldoffsetlocks;
  63     //private Vector<ReentrantLock> heldblocklocks;
  64     private Vector heldblocklocks;
  65     //private HashMap<INode, Vector<TransactionalFile>> AccessedFiles;
  66     private HashMap AccessedFiles;
  67     //private HashMap<INode, HashMap<Integer, BlockAccessModesEnum> > accessedBlocks;
  68     private HashMap accessedBlocks;
  69     //private HashMap<TransactionalFile, TransactionLocalFileAttributes> LocaltoGlobalMappings;
  70     private HashMap GlobaltoLocalMappings;
  71     public HashMap merge_for_writes_done;
  72     private HashMap writeBuffer;
  73     private ContentionManager contentionmanager;
  74     private /*volatile*/ Status status;
  75     private int id;
  76
  77     public ExtendedTransaction() {
  78         //  super();
  79         // id = Integer.valueOf(Thread.currentThread().getName().substring(7));
  80         heldblocklocks = new Vector();
  81         heldoffsetlocks = new Vector();
  82         AccessedFiles = new HashMap();
  83         GlobaltoLocalMappings = new HashMap/*<TransactionalFile, TransactionLocalFileAttributes >*/();
  84         writeBuffer = new HashMap();
  85         status = Status.ACTIVE;
  86         accessedBlocks = new HashMap();
  87         merge_for_writes_done = new HashMap();
  88         writesmerged = true;
  89     //   setContentionmanager(new BaseManager());
  90     //    beginTransaction();
  91
  92     }
  93
  94     public ExtendedTransaction(TransactionStatu memorystate) {
  95         this();
  96
  97         this.memorystate = memorystate;
  98     }
  99
 100     private int invokeNativepwrite(byte buff[], long offset, int size, RandomAccessFile file) {
 101         try {
 102             return nativepwrite(buff, offset, buff.length, file.getFD());
 103         } catch (IOException ex) {
 104
 105             Logger.getLogger(TransactionalFile.class.getName()).log(Level.SEVERE, null, ex);
 106             return -1;
 107         }
 108
 109     }
 110
 111     public void beginTransaction() {
 112         this.addPropertyChangeListener(new customhandler(Status.ABORTED));
 113     }
 114
 115     public void abort() {
 116         synchronized (this) {
 117
 118             this.status = Status.ABORTED;
 119             if (this.memorystate != null && !(this.memorystate).isAborted()) {
 120                 this.memorystate.abortThisSystem();
 121             }
 122
 123         }
 124
 125     }
 126
 127     public Status getStatus() {
 128         return status;
 129     }
 130
 131     public boolean isActive() {
 132         return this.getStatus() == Status.ACTIVE;
 133     }
 134
 135     public boolean isAborted() {
 136         return this.getStatus() == Status.ABORTED;
 137     }
 138
 139     public ContentionManager getContentionmanager() {
 140         return contentionmanager;
 141     }
 142
 143     public void setContentionmanager(ContentionManager contentionmanager) {
 144         this.contentionmanager = contentionmanager;
 145     }
 146
 147     public HashMap getWriteBuffer() {
 148         return writeBuffer;
 149     }
 150
 151     public HashMap getAccessedFiles() {
 152         return AccessedFiles;
 153     }
 154
 155     public boolean isWritesmerged() {
 156         return writesmerged;
 157     }
 158
 159     public void setWritesmerged(boolean writesmerged) {
 160         this.writesmerged = writesmerged;
 161     }
 162
 163     public HashMap getGlobaltoLocalMappings() {
 164         return GlobaltoLocalMappings;
 165     }
 166
 167     public HashMap getAccessedBlocks() {
 168         return accessedBlocks;
 169     }
 170
 171     public ContentionManager getBlockContentionManager() {
 172         return ManagerRepository.getBlockcm();
 173     }
 174
 175     public ContentionManager getOffsetContentionManager() {
 176         return ManagerRepository.getOffsetcm();
 177     }
 178
 179     public TreeMap getSortedFileAccessMap(HashMap hmap) {
 180         /*TreeMap sortedMap = new TreeMap(hmap);
 181         return sortedMap;*/
 182         return new TreeMap(hmap);
 183     }
 184
 185     public void setStatus(Status st) {
 186         Status oldst = getStatus();
 187         this.status = st;
 188         this.changes.firePropertyChange("status", oldst, st);
 189     }
 190
 191     public void addFile(TransactionalFile tf, long offsetnumber/*, TransactionLocalFileAttributes tmp*/) {
 192
 193         TransactionLocalFileAttributes tmp = new TransactionLocalFileAttributes(offsetnumber/*, tf.getInodestate().commitedfilesize.get()*/);
 194         Vector dummy;
 195
 196         if (AccessedFiles.containsKey(tf.getInode())) {
 197             dummy = (Vector) AccessedFiles.get(tf.getInode());
 198         } else {
 199             dummy = new Vector();
 200             AccessedFiles.put(tf.getInode(), dummy);
 201         }
 202         dummy.add(tf);
 203         GlobaltoLocalMappings.put(tf, tmp);
 204         merge_for_writes_done.put(tf.getInode(), Boolean.TRUE);
 205     }
 206
 207     public boolean lockOffsets() {   /// Locking offsets for File Descriptors
 208
 209
 210         TreeMap hm = getSortedFileAccessMap(AccessedFiles);
 211         Iterator iter = hm.keySet().iterator();
 212
 213         while (iter.hasNext() && (this.getStatus() == Status.ACTIVE)) {
 214             INode key = (INode) iter.next();
 215
 216             Vector vec = (Vector) AccessedFiles.get(key);
 217             Collections.sort(vec);
 218             Iterator it = vec.iterator();
 219             while (it.hasNext()) {
 220                 TransactionalFile value = (TransactionalFile) it.next();
 221                 while (this.getStatus() == Status.ACTIVE) {
 222                     value.offsetlock.lock();
 223
 224                     heldoffsetlocks.add(value.offsetlock);
 225                     break;
 226
 227                 }
 228                 if (this.getStatus() != Status.ACTIVE) {
 229
 230
 231                     return false;
 232                 }
 233             }
 234
 235         }
 236         if (this.getStatus() != Status.ACTIVE) {
 237
 238
 239             return false;
 240         }
 241         return true;
 242     }
 243
 244     public boolean lockBlock(BlockDataStructure block, BlockAccessModesEnum mode/*, GlobalINodeState adapter, BlockAccessModesEnum mode, int expvalue, INode inode, TransactionLocalFileAttributes tf*/) {
 245
 246         Lock lock;
 247         if (mode == BlockAccessModesEnum.READ) {
 248             lock = block.getLock().readLock();
 249         } else {
 250
 251             lock = block.getLock().writeLock();
 252         }
 253
 254         while (this.getStatus() == Status.ACTIVE) {
 255             lock.lock();
 256             heldblocklocks.add(lock);
 257             return true;
 258         }
 259         return false;
 260     }
 261
 262     public void prepareCommit() {
 263         if (this.status != Status.ACTIVE) {
 264             throw new AbortedException();
 265         }
 266         boolean ok = true;
 267         if (!lockOffsets()) {
 268             throw new AbortedException();
 269         }
 270
 271
 272         ///////////////////////////
 273
 274
 275         Map hm = getWriteBuffer();
 276
 277         Iterator iter = hm.keySet().iterator();
 278         WriteOperations value;
 279         Vector vec = new Vector();
 280         while (iter.hasNext() && (this.getStatus() == Status.ACTIVE) && ok) {
 281             INode key = (INode) iter.next();
 282             vec = (Vector) hm.get(key);
 283             Collections.sort(vec);
 284             Iterator it = vec.iterator();
 285             while (it.hasNext()) {
 286
 287                 value = (WriteOperations) it.next();
 288                 if (value.isUnknownoffset()) {
 289
 290                     long start;
 291                     long end;
 292                     start = value.getRange().getStart() - value.getBelongingto().getCopylocaloffset() + value.getOwnertransactionalFile().getCommitedoffset().getOffsetnumber();
 293                     end = value.getRange().getEnd() - value.getBelongingto().getCopylocaloffset() + value.getOwnertransactionalFile().getCommitedoffset().getOffsetnumber();
 294                     if (value.getBelongingto().isUnknown_inital_offset_for_write()) {
 295                         value.getBelongingto().setLocaloffset(value.getBelongingto().getLocaloffset() - value.getBelongingto().getCopylocaloffset() + value.getOwnertransactionalFile().getCommitedoffset().getOffsetnumber());
 296                         value.getBelongingto().setUnknown_inital_offset_for_write(false);
 297                     }
 298
 299                     int startblock = FileBlockManager.getCurrentFragmentIndexofTheFile(start);
 300                     int targetblock = FileBlockManager.getTargetFragmentIndexofTheFile(start, value.getRange().getEnd() - value.getRange().getStart());
 301
 302                     TreeMap sset;
 303                     if (this.getAccessedBlocks().get(key) != null) {
 304                         sset = (TreeMap) this.getAccessedBlocks().get(key);
 305                     } else {
 306                         sset = new TreeMap();
 307                         this.getAccessedBlocks().put(key, sset);
 308                     }
 309
 310
 311                     for (int i = startblock; i <= targetblock; i++) {
 312                         if (sset.containsKey(Integer.valueOf(i))) {
 313                             if (sset.get(Integer.valueOf(i)) != BlockAccessModesEnum.WRITE) {
 314                                 sset.put(Integer.valueOf(i), BlockAccessModesEnum.READ_WRITE);
 315                             }
 316                         } else {
 317                             sset.put(Integer.valueOf(i), BlockAccessModesEnum.WRITE);
 318                         }
 319                     }
 320
 321                     value.getRange().setStart(start);
 322                     value.getRange().setEnd(end);
 323                 }
 324             }
 325
 326         }
 327
 328         Iterator it = this.getAccessedBlocks().keySet().iterator();
 329         while (it.hasNext() && (this.getStatus() == Status.ACTIVE)) {
 330             INode inode = (INode) it.next();
 331             GlobalINodeState inodestate = TransactionalFileWrapperFactory.getTateransactionalFileINodeState(inode);
 332             TreeMap vec2 = (TreeMap) this.getAccessedBlocks().get(inode);
 333             Iterator iter2 = vec2.keySet().iterator();
 334             while (iter2.hasNext()) {
 335                 Integer num = (Integer) iter2.next();
 336
 337                 BlockDataStructure blockobj = inodestate.getBlockDataStructure(num);
 338
 339                 ok = this.lockBlock(blockobj, (BlockAccessModesEnum) vec2.get(num));
 340                 if (ok == false) {
 341                     break;
 342                 }
 343             }
 344         }
 345
 346         if (this.getStatus() != Status.ACTIVE) {
 347
 348             throw new AbortedException();
 349         }
 350         abortAllReaders();
 351
 352     }
 353
 354     public void commitChanges() {
 355
 356         Map hm = getWriteBuffer();
 357         Iterator iter = hm.keySet().iterator();
 358         Iterator it;
 359         WriteOperations writeop;
 360         Vector vec;
 361         while (iter.hasNext() && (this.getStatus() == Status.ACTIVE)) {
 362             INode key = (INode) iter.next();
 363
 364             vec = (Vector) hm.get(key);
 365             Collections.sort(vec);
 366             it = vec.iterator();
 367             while (it.hasNext()) {
 368
 369                 writeop = (WriteOperations) it.next();
 370                 Byte[] data = new Byte[(int) (writeop.getRange().getEnd() - writeop.getRange().getStart())];
 371                 byte[] bytedata = new byte[(int) (writeop.getRange().getEnd() - writeop.getRange().getStart())];
 372                 data = (Byte[]) writeop.getData();
 373
 374                 for (int i = 0; i < data.length; i++) {
 375                     bytedata[i] = data[i];
 376                 }
 377                 invokeNativepwrite(bytedata, writeop.getRange().getStart(), bytedata.length, writeop.getOwnertransactionalFile().file);
 378             }
 379         }
 380
 381         Iterator k = GlobaltoLocalMappings.keySet().iterator();
 382         while (k.hasNext()) {
 383             TransactionalFile trf = (TransactionalFile) (k.next());
 384             trf.getCommitedoffset().setOffsetnumber(((TransactionLocalFileAttributes) GlobaltoLocalMappings.get(trf)).getLocaloffset());
 385         }
 386     }
 387
 388     public void unlockAllLocks() {
 389         Iterator it = heldblocklocks.iterator();
 390
 391         while (it.hasNext()) {
 392
 393             Lock lock = (Lock) it.next();
 394             lock.unlock();
 395         }
 396         heldblocklocks.clear();
 397
 398         it = heldoffsetlocks.iterator();
 399         while (it.hasNext()) {
 400             ReentrantLock lock = (ReentrantLock) it.next();
 401             lock.unlock();
 402         }
 403         heldoffsetlocks.clear();
 404     }
 405
 406     public void abortAllReaders() {
 407         TreeMap hm = getSortedFileAccessMap(AccessedFiles);
 408         //lock phase
 409         Iterator iter = hm.keySet().iterator();
 410         TransactionalFile value;
 411         while (iter.hasNext()) {
 412             INode key = (INode) iter.next();
 413             Vector vec = (Vector) AccessedFiles.get(key);
 414             Iterator it = vec.iterator();
 415             while (it.hasNext()) {
 416
 417                 value = (TransactionalFile) it.next();
 418                 Iterator it2 = value.getCommitedoffset().getOffsetReaders().iterator(); // for visible readers strategy
 419
 420                 while (it2.hasNext()) {
 421                     ExtendedTransaction tr = (ExtendedTransaction) it2.next();
 422                     if (tr != this) {
 423                         tr.abort();
 424                     }
 425                 }
 426                 value.getCommitedoffset().getOffsetReaders().clear();
 427             }
 428
 429             TreeMap vec2;
 430             if (accessedBlocks.get(key) != null) {
 431                 vec2 = (TreeMap) accessedBlocks.get(key);
 432             } else {
 433                 vec2 = new TreeMap();
 434
 435             }
 436             GlobalINodeState inodestate = TransactionalFileWrapperFactory.getTateransactionalFileINodeState(key);
 437             Iterator it2 = vec2.keySet().iterator();
 438
 439             while (it2.hasNext()) {
 440
 441                 Integer num = (Integer) it2.next();
 442                 if (vec2.get(num) != BlockAccessModesEnum.READ) {
 443                     BlockDataStructure blockobj = (BlockDataStructure) inodestate.getBlockDataStructure(num);//lockmap.get(num);
 444
 445                     Iterator it4 = blockobj.getReaders().iterator(); // from here for visible readers strategy
 446
 447                     while (it4.hasNext()) {
 448
 449                         ExtendedTransaction tr = (ExtendedTransaction) it4.next();
 450                         if (this != tr) {
 451                             tr.abort();
 452                         }
 453                     }
 454                     blockobj.getReaders().clear();
 455
 456                 }
 457             }
 458
 459
 460         }
 461     }
 462
 463     public void addPropertyChangeListener(PropertyChangeListener listener) {
 464         this.changes.addPropertyChangeListener("status", listener);
 465     }
 466
 467     public void removePropertyChangeListener(PropertyChangeListener listener) {
 468         this.changes.removePropertyChangeListener("status", listener);
 469     }
 470
 471     public TransactionStatu getOtherSystem() {
 472         return memorystate;
 473     }
 474
 475     public void setOtherSystem(TransactionStatu othersystem) {
 476         memorystate = othersystem;
 477     }
 478
 479     public Vector getHeldblocklocks() {
 480         return heldblocklocks;
 481     }
 482
 483     public void setHeldblocklocks(Vector heldblocklocks) {
 484         this.heldblocklocks = heldblocklocks;
 485     }
 486
 487     public Vector getHeldoffsetlocks() {
 488         return heldoffsetlocks;
 489     }
 490
 491     public void setHeldoffsetlocks(Vector heldoffsetlocks) {
 492         this.heldoffsetlocks = heldoffsetlocks;
 493     }
 494
 495     public void abortThisSystem() {
 496         abort();
 497     }
 498
 499     public boolean isCommitted() {
 500         if (this.status == Status.COMMITTED) {
 501             return true;
 502         }
 503         return false;
 504
 505     }
 506 }
 507 /*
 508 public boolean lockBlock(BlockDataStructure block, Adapter adapter, BlockAccessModesEnum mode, int expvalue) { // from here for visible readers strategy
 509 while (this.getStatus() == Status.ACTIVE) {
 510 if (lock.tryLock()) {
 511 Thread.onAbortOnce(new Runnable() {
 512
 513 public void run() {
 514 lock.unlock();
 515 }
 516 });
 517
 518 heldblocklocks.add(lock);
 519
 520 synchronized (adapter) {
 521 block.setOwner(this);
 522 //        Iterator it =  block.getReaders().iterator();
 523 //        while (it.hasNext())
 524 //        {
 525 //            ExtendedTransaction tr = (ExtendedTransaction) it.next();
 526 //            tr.abort();
 527 //       }
 528 }
 529
 530 return true;
 531 } else {
 532 getBlockContentionManager().resolveConflict(this, block.getOwner());
 533 }
 534 }
 535 return false;*/
 536 /*
 537 public boolean lockBlock(BlockDataStructure block, Adapter adapter, BlockAccessModesEnum mode, int expvalue) { // versioning strat
 538 while (this.getStatus() == Status.ACTIVE) {
 539 if (lock.tryLock()) {
 540 Thread.onAbortOnce(new Runnable() {
 541
 542 public void run() {
 543 lock.unlock();
 544 }
 545 });
 546
 547 heldblocklocks.add(lock);
 548 if (mode != BlockAccessModesEnum.WRITE) {   egy
 549 if (block.getVersion().get() != expvalue) {
 550 unlockAllLocks();
 551 return false;
 552 }
 553 }
 554 synchronized (adapter) {
 555 block.setOwner(this);
 556 }
 557
 558 return true;
 559 } else {
 560 getContentionManager().resolveConflict(this, block.getOwner());
 561 }
 562 }
 563 return false;
 564 }*/
 565 // }
 566 //expvalue = ((Integer) value.getBlockversions().get(it)).intValue(); //for versioning strategy
 567             /*if (!(value.isValidatelocaloffset())) {
 568 if (((BlockAccessModesEnum) (value.getAccesedblocks().get(blockno))) != BlockAccessModesEnum.WRITE) { //versioning strategy
 569
 570 /if (blockobj.getVersion().get() == expvalue) {
 571
 572 ok = this.lock(blockobj, value.adapter, (BlockAccessModesEnum) (value.getAccesedblocks().get(blockno)), expvalue);
 573 if (ok == false) {
 574 //        unlockAllLocks();
 575 break;
 576 }
 577 } else {
 578 ok = false;
 579 break;
 580 }
 581 } else {
 582
 583 ok = this.lock(blockobj, value.adapter, (BlockAccessModesEnum) (value.getAccesedblocks().get(blockno)), expvalue);
 584 if (ok == false) {
 585 break;
 586 }
 587 }
 588 }
 589
 590
 591 if (!(ok)) {
 592 unlockAllLocks();
 593 throw new AbortedException();
 594 }*/
 595
 596
 597
 598