package org.apache.lucene.index;

/**
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.lucene.store.IndexInput;
import org.apache.lucene.store.IndexOutput;

/**
 * Byte[] referencing is used because a new norm object needs 
 * to be created for each clone, and the byte array is all 
 * that is needed for sharing between cloned readers.  The 
 * current norm referencing is for sharing between readers 
 * whereas the byte[] referencing is for copy on write which 
 * is independent of reader references (i.e. incRef, decRef).
 */

final class SegmentNorms implements Cloneable {

  /** norms header placeholder */
  static final byte[] NORMS_HEADER = new byte[]{'N','R','M',-1};

  int refCount = 1;

  // If this instance is a clone, the originalNorm
  // references the Norm that has a real open IndexInput:
  private SegmentNorms origNorm;

  private IndexInput in;
  private long normSeek;

  // null until bytes is set
  private AtomicInteger bytesRef;
  private byte[] bytes;
  private int number;

  boolean dirty;
  boolean rollbackDirty;
  
  private final SegmentReader owner;
  
  public SegmentNorms(IndexInput in, int number, long normSeek, SegmentReader owner) {
    this.in = in;
    this.number = number;
    this.normSeek = normSeek;
    this.owner = owner;
  }

  public synchronized void incRef() {
    assert refCount > 0 && (origNorm == null || origNorm.refCount > 0);
    refCount++;
  }

  private void closeInput() throws IOException {
    if (in != null) {
      if (in != owner.singleNormStream) {
        // It's private to us -- just close it
        in.close();
      } else {
        // We are sharing this with others -- decRef and
        // maybe close the shared norm stream
        if (owner.singleNormRef.decrementAndGet() == 0) {
          owner.singleNormStream.close();
          owner.singleNormStream = null;
        }
      }

      in = null;
    }
  }

  public synchronized void decRef() throws IOException {
    assert refCount > 0 && (origNorm == null || origNorm.refCount > 0);

    if (--refCount == 0) {
      if (origNorm != null) {
        origNorm.decRef();
        origNorm = null;
      } else {
        closeInput();
      }

      if (bytes != null) {
        assert bytesRef != null;
        bytesRef.decrementAndGet();
        bytes = null;
        bytesRef = null;
      } else {
        assert bytesRef == null;
      }
    }
  }

  // Load bytes but do not cache them if they were not
  // already cached
  public synchronized void bytes(byte[] bytesOut, int offset, int len) throws IOException {
    assert refCount > 0 && (origNorm == null || origNorm.refCount > 0);
    if (bytes != null) {
      // Already cached -- copy from cache:
      assert len <= owner.maxDoc();
      System.arraycopy(bytes, 0, bytesOut, offset, len);
    } else {
      // Not cached
      if (origNorm != null) {
        // Ask origNorm to load
        origNorm.bytes(bytesOut, offset, len);
      } else {
        // We are orig -- read ourselves from disk:
        synchronized(in) {
          in.seek(normSeek);
          in.readBytes(bytesOut, offset, len, false);
        }
      }
    }
  }

  // Load & cache full bytes array.  Returns bytes.
  public synchronized byte[] bytes() throws IOException {
    assert refCount > 0 && (origNorm == null || origNorm.refCount > 0);
    if (bytes == null) {                     // value not yet read
      assert bytesRef == null;
      if (origNorm != null) {
        // Ask origNorm to load so that for a series of
        // reopened readers we share a single read-only
        // byte[]
        bytes = origNorm.bytes();
        bytesRef = origNorm.bytesRef;
        bytesRef.incrementAndGet();

        // Once we've loaded the bytes we no longer need
        // origNorm:
        origNorm.decRef();
        origNorm = null;

      } else {
        // We are the origNorm, so load the bytes for real
        // ourself:
        final int count = owner.maxDoc();
        bytes = new byte[count];

        // Since we are orig, in must not be null
        assert in != null;

        // Read from disk.
        synchronized(in) {
          in.seek(normSeek);
          in.readBytes(bytes, 0, count, false);
        }

        bytesRef = new AtomicInteger(1);
        closeInput();
      }
    }

    return bytes;
  }

  // Only for testing
  AtomicInteger bytesRef() {
    return bytesRef;
  }

  // Called if we intend to change a norm value.  We make a
  // private copy of bytes if it's shared with others:
  public synchronized byte[] copyOnWrite() throws IOException {
    assert refCount > 0 && (origNorm == null || origNorm.refCount > 0);
    bytes();
    assert bytes != null;
    assert bytesRef != null;
    if (bytesRef.get() > 1) {
      // I cannot be the origNorm for another norm
      // instance if I'm being changed.  Ie, only the
      // "head Norm" can be changed:
      assert refCount == 1;
      final AtomicInteger oldRef = bytesRef;
      bytes = owner.cloneNormBytes(bytes);
      bytesRef = new AtomicInteger(1);
      oldRef.decrementAndGet();
    }
    dirty = true;
    return bytes;
  }
  
  // Returns a copy of this Norm instance that shares
  // IndexInput & bytes with the original one
  @Override
  public synchronized Object clone() {
    assert refCount > 0 && (origNorm == null || origNorm.refCount > 0);
      
    SegmentNorms clone;
    try {
      clone = (SegmentNorms) super.clone();
    } catch (CloneNotSupportedException cnse) {
      // Cannot happen
      throw new RuntimeException("unexpected CloneNotSupportedException", cnse);
    }
    clone.refCount = 1;

    if (bytes != null) {
      assert bytesRef != null;
      assert origNorm == null;

      // Clone holds a reference to my bytes:
      clone.bytesRef.incrementAndGet();
    } else {
      assert bytesRef == null;
      if (origNorm == null) {
        // I become the origNorm for the clone:
        clone.origNorm = this;
      }
      clone.origNorm.incRef();
    }

    // Only the origNorm will actually readBytes from in:
    clone.in = null;

    return clone;
  }

  // Flush all pending changes to the next generation
  // separate norms file.
  public void reWrite(SegmentInfo si) throws IOException {
    assert refCount > 0 && (origNorm == null || origNorm.refCount > 0): "refCount=" + refCount + " origNorm=" + origNorm;

    // NOTE: norms are re-written in regular directory, not cfs
    si.advanceNormGen(this.number);
    final String normFileName = si.getNormFileName(this.number);
    IndexOutput out = owner.directory().createOutput(normFileName);
    boolean success = false;
    try {
      try {
        out.writeBytes(SegmentNorms.NORMS_HEADER, 0, SegmentNorms.NORMS_HEADER.length);
        out.writeBytes(bytes, owner.maxDoc());
      } finally {
        out.close();
      }
      success = true;
    } finally {
      if (!success) {
        try {
          owner.directory().deleteFile(normFileName);
        } catch (Throwable t) {
          // suppress this so we keep throwing the
          // original exception
        }
      }
    }
    this.dirty = false;
  }
}