1/*  Part of SWI-Prolog
    2
    3    Author:        Jan Wielemaker
    4    E-mail:        J.Wielemaker@vu.nl
    5    WWW:           http://www.swi-prolog.org
    6    Copyright (c)  2006-2015, University of Amsterdam
    7                              VU University Amsterdam
    8    All rights reserved.
    9
   10    Redistribution and use in source and binary forms, with or without
   11    modification, are permitted provided that the following conditions
   12    are met:
   13
   14    1. Redistributions of source code must retain the above copyright
   15       notice, this list of conditions and the following disclaimer.
   16
   17    2. Redistributions in binary form must reproduce the above copyright
   18       notice, this list of conditions and the following disclaimer in
   19       the documentation and/or other materials provided with the
   20       distribution.
   21
   22    THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
   23    "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
   24    LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS
   25    FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE
   26    COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
   27    INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
   28    BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
   29    LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
   30    CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
   31    LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN
   32    ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
   33    POSSIBILITY OF SUCH DAMAGE.
   34*/
   35
   36:- module(rdf_persistency,
   37          [ rdf_attach_db/2,            % +Directory, +Options
   38            rdf_detach_db/0,            % +Detach current Graph
   39            rdf_current_db/1,           % -Directory
   40            rdf_persistency/2,          % +Graph, +Bool
   41            rdf_flush_journals/1,       % +Options
   42            rdf_persistency_property/1, % ?Property
   43            rdf_journal_file/2,         % ?Graph, ?JournalFile
   44            rdf_snapshot_file/2,        % ?Graph, ?SnapshotFile
   45            rdf_db_to_file/2            % ?Graph, ?FileBase
   46          ]).   47:- use_module(library(semweb/rdf_db),
   48              [ rdf_graph/1, rdf_unload_graph/1, rdf_statistics/1,
   49                rdf_load_db/1, rdf_retractall/4, rdf_create_graph/1,
   50                rdf_assert/4, rdf_update/5, rdf_monitor/2, rdf/4,
   51                rdf_save_db/2, rdf_atom_md5/3, rdf_current_ns/2,
   52                rdf_register_ns/3
   53              ]).   54
   55:- autoload(library(apply),[maplist/2,maplist/3,partition/4,exclude/3]).   56:- use_module(library(debug),[debug/3]).   57:- autoload(library(error),
   58	    [permission_error/3,must_be/2,domain_error/2]).   59:- autoload(library(filesex),
   60	    [directory_file_path/3,make_directory_path/1]).   61:- autoload(library(lists),[select/3,append/3]).   62:- autoload(library(option),[option/2,option/3]).   63:- autoload(library(readutil),[read_file_to_terms/3]).   64:- autoload(library(socket),[gethostname/1]).   65:- autoload(library(thread),[concurrent/3]).   66:- autoload(library(uri),[uri_encoded/3]).   67
   68/** <module> RDF persistency plugin
   69
   70This  module  provides  persistency   for    rdf_db.pl   based   on  the
   71rdf_monitor/2 predicate to  track  changes   to  the  repository.  Where
   72previous  versions  used  autosafe  of  the  whole  database  using  the
   73quick-load format of rdf_db, this version is  based on a quick-load file
   74per source (4th argument of rdf/4), and journalling for edit operations.
   75
   76The result is safe, avoids frequent small   changes to large files which
   77makes synchronisation and backup expensive and avoids long disruption of
   78the server doing the autosafe. Only loading large files disrupts service
   79for some time.
   80
   81The persistent backup of the database is  realised in a directory, using
   82a lock file to avoid corruption due to concurrent access. Each source is
   83represented by two files, the latest snapshot   and a journal. The state
   84is restored by loading  the  snapshot   and  replaying  the journal. The
   85predicate rdf_flush_journals/1 can be used to create fresh snapshots and
   86delete the journals.
   87
   88@tbd If there is a complete `.new'   snapshot  and no journal, we should
   89     move the .new to the plain snapshot name as a means of recovery.
   90
   91@tbd Backup of each graph using one or two files is very costly if there
   92     are many graphs.  Although the currently used subdirectories avoid
   93     hitting OS limits early, this is still not ideal. Probably we
   94     should collect (small, older?) files and combine them into a single
   95     quick load file.  We could call this (similar to GIT) a `pack'.
   96
   97@see    rdf_edit.pl
   98*/
   99
  100:- volatile
  101    rdf_directory/1,
  102    rdf_lock/2,
  103    rdf_option/1,
  104    source_journal_fd/2,
  105    file_base_db/2.  106:- dynamic
  107    rdf_directory/1,                % Absolute path
  108    rdf_lock/2,                     % Dir, Lock
  109    rdf_option/1,                   % Defined options
  110    source_journal_fd/2,            % DB, JournalFD
  111    file_base_db/2.                 % FileBase, DB
  112
  113:- meta_predicate
  114    no_agc(0).  115
  116:- predicate_options(rdf_attach_db/2, 2,
  117                     [ access(oneof([read_write,read_only])),
  118                       concurrency(positive_integer),
  119                       max_open_journals(positive_integer),
  120                       silent(oneof([true,false,brief])),
  121                       log_nested_transactions(boolean)
  122                     ]).  123
  124%!  rdf_attach_db(+Directory, +Options) is det.
  125%
  126%   Start persistent operations using Directory   as  place to store
  127%   files.   There are several cases:
  128%
  129%           * Empty DB, existing directory
  130%           Load the DB from the existing directory
  131%
  132%           * Full DB, empty directory
  133%           Create snapshots for all sources in directory
  134%
  135%   Options:
  136%
  137%           * access(+AccessMode)
  138%           One of =auto= (default), =read_write= or
  139%           =read_only=. Read-only access implies that the RDF
  140%           store is not locked. It is read at startup and all
  141%           modifications to the data are temporary. The default
  142%           =auto= mode is =read_write= if the directory is
  143%           writeable and the lock can be acquired.  Otherwise
  144%           it reverts to =read_only=.
  145%
  146%           * concurrency(+Jobs)
  147%           Number of threads to use for loading the initial
  148%           database.  If not provided it is the number of CPUs
  149%           as obtained from the flag =cpu_count=.
  150%
  151%           * max_open_journals(+Count)
  152%           Maximum number of journals kept open.  If not provided,
  153%           the default is 10.  See limit_fd_pool/0.
  154%
  155%           * directory_levels(+Count)
  156%           Number of levels of intermediate directories for storing
  157%           the graph files.  Default is 2.
  158%
  159%           * silent(+BoolOrBrief)
  160%           If =true= (default =false=), do not print informational
  161%           messages.  Finally, if =brief= it will show minimal
  162%           feedback.
  163%
  164%           * log_nested_transactions(+Boolean)
  165%           If =true=, nested _log_ transactions are added to the
  166%           journal information.  By default (=false=), no log-term
  167%           is added for nested transactions.\\
  168%
  169%   @error existence_error(source_sink, Directory)
  170%   @error permission_error(write, directory, Directory)
  171
  172rdf_attach_db(DirSpec, Options) :-
  173    option(access(read_only), Options),
  174    !,
  175    absolute_file_name(DirSpec,
  176                       Directory,
  177                       [ access(read),
  178                         file_type(directory)
  179                       ]),
  180    rdf_attach_db_ro(Directory, Options).
  181rdf_attach_db(DirSpec, Options) :-
  182    option(access(read_write), Options),
  183    !,
  184    rdf_attach_db_rw(DirSpec, Options).
  185rdf_attach_db(DirSpec, Options) :-
  186    absolute_file_name(DirSpec,
  187                       Directory,
  188                       [ access(exist),
  189                         file_type(directory),
  190                         file_errors(fail)
  191                       ]),
  192    !,
  193    (   access_file(Directory, write)
  194    ->  catch(rdf_attach_db_rw(Directory, Options), E, true),
  195        (   var(E)
  196        ->  true
  197        ;   E = error(permission_error(lock, rdf_db, _), _)
  198        ->  print_message(warning, E),
  199            print_message(warning, rdf(read_only)),
  200            rdf_attach_db(DirSpec, [access(read_only)|Options])
  201        ;   throw(E)
  202        )
  203    ;   print_message(warning,
  204                      error(permission_error(write, directory, Directory))),
  205        print_message(warning, rdf(read_only)),
  206        rdf_attach_db_ro(Directory, Options)
  207    ).
  208rdf_attach_db(DirSpec, Options) :-
  209    catch(rdf_attach_db_rw(DirSpec, Options), E, true),
  210    (   var(E)
  211    ->  true
  212    ;   print_message(warning, E),
  213        print_message(warning, rdf(read_only)),
  214        rdf_attach_db(DirSpec, [access(read_only)|Options])
  215    ).
  216
  217
  218rdf_attach_db_rw(DirSpec, Options) :-
  219    absolute_file_name(DirSpec,
  220                       Directory,
  221                       [ access(write),
  222                         file_type(directory),
  223                         file_errors(fail)
  224                       ]),
  225    !,
  226    (   rdf_directory(Directory)
  227    ->  true                        % update settings?
  228    ;   rdf_detach_db,
  229        mkdir(Directory),
  230        lock_db(Directory),
  231        assert(rdf_directory(Directory)),
  232        assert_options(Options),
  233        stop_monitor,               % make sure not to register load
  234        no_agc(load_db),
  235        at_halt(rdf_detach_db),
  236        start_monitor
  237    ).
  238rdf_attach_db_rw(DirSpec, Options) :-
  239    absolute_file_name(DirSpec,
  240                       Directory,
  241                       [ solutions(all)
  242                       ]),
  243    (   exists_directory(Directory)
  244    ->  access_file(Directory, write)
  245    ;   catch(make_directory(Directory), _, fail)
  246    ),
  247    !,
  248    rdf_attach_db(Directory, Options).
  249rdf_attach_db_rw(DirSpec, _) :-         % Generate an existence or
  250    absolute_file_name(DirSpec,     % permission error
  251                       Directory,
  252                       [ access(exist),
  253                         file_type(directory)
  254                       ]),
  255    permission_error(write, directory, Directory).
  256
  257%!  rdf_attach_db_ro(+Directory, +Options)
  258%
  259%   Open an RDF database in read-only mode.
  260
  261rdf_attach_db_ro(Directory, Options) :-
  262    rdf_detach_db,
  263    assert(rdf_directory(Directory)),
  264    assert_options(Options),
  265    stop_monitor,           % make sure not to register load
  266    no_agc(load_db).
  267
  268
  269assert_options([]).
  270assert_options([H|T]) :-
  271    (   option_type(H, Check)
  272    ->  Check,
  273        assert(rdf_option(H))
  274    ;   true                        % ignore options we do not understand
  275    ),
  276    assert_options(T).
  277
  278option_type(concurrency(X),             must_be(positive_integer, X)).
  279option_type(max_open_journals(X),       must_be(positive_integer, X)).
  280option_type(directory_levels(X),        must_be(positive_integer, X)).
  281option_type(silent(X),                  must_be(oneof([true,false,brief]), X)).
  282option_type(log_nested_transactions(X), must_be(boolean, X)).
  283option_type(access(X),                  must_be(oneof([read_write,
  284                                                       read_only]), X)).
  285
  286
  287%!  rdf_persistency_property(?Property) is nondet.
  288%
  289%   True when Property is a property of the current persistent database.
  290%   Exposes  the  properties  that  can   be    passed   as  options  to
  291%   rdf_attach_db/2.                                       Specifically,
  292%   rdf_persistency_property(access(read_only)) is true iff the database
  293%   is mounted in read-only mode. In addition, the following property is
  294%   supported:
  295%
  296%     - directory(Dir)
  297%     The directory in which the database resides.
  298
  299rdf_persistency_property(Property) :-
  300    var(Property),
  301    !,
  302    rdf_persistency_property_(Property).
  303rdf_persistency_property(Property) :-
  304    rdf_persistency_property_(Property),
  305    !.
  306
  307rdf_persistency_property_(Property) :-
  308    rdf_option(Property).
  309rdf_persistency_property_(directory(Dir)) :-
  310    rdf_directory(Dir).
  311
  312%!  no_agc(:Goal)
  313%
  314%   Run Goal with atom garbage collection   disabled. Loading an RDF
  315%   database creates large amounts  of  atoms   we  *know*  are  not
  316%   garbage.
  317
  318no_agc(Goal) :-
  319    setup_call_cleanup(
  320        push_prolog_flag(agc_margin, 0),
  321        Goal,
  322        pop_prolog_flag(agc_margin)).
  323
  324
  325%!  rdf_detach_db is det.
  326%
  327%   Detach from the  current  database.   Succeeds  silently  if  no
  328%   database is attached. Normally called at  the end of the program
  329%   through at_halt/1.
  330
  331rdf_detach_db :-
  332    debug(halt, 'Detaching RDF database', []),
  333    stop_monitor,
  334    close_journals,
  335    (   retract(rdf_directory(Dir))
  336    ->  debug(halt, 'DB Directory: ~w', [Dir]),
  337        save_prefixes(Dir),
  338        retractall(rdf_option(_)),
  339        retractall(source_journal_fd(_,_)),
  340        unlock_db(Dir)
  341    ;   true
  342    ).
  343
  344
  345%!  rdf_current_db(?Dir)
  346%
  347%   True if Dir is the current RDF persistent database.
  348
  349rdf_current_db(Directory) :-
  350    rdf_directory(Dir),
  351    !,
  352    Dir = Directory.
  353
  354
  355%!  rdf_flush_journals(+Options)
  356%
  357%   Flush dirty journals.  Options:
  358%
  359%           * min_size(+KB)
  360%           Only flush if journal is over KB in size.
  361%           * graph(+Graph)
  362%           Only flush the journal of Graph
  363%
  364%   @tbd Provide a default for min_size?
  365
  366rdf_flush_journals(Options) :-
  367    option(graph(Graph), Options, _),
  368    forall(rdf_graph(Graph),
  369           rdf_flush_journal(Graph, Options)).
  370
  371rdf_flush_journal(Graph, Options) :-
  372    db_files(Graph, _SnapshotFile, JournalFile),
  373    db_file(JournalFile, File),
  374    (   \+ exists_file(File)
  375    ->  true
  376    ;   memberchk(min_size(KB), Options),
  377        size_file(File, Size),
  378        Size / 1024 < KB
  379    ->  true
  380    ;   create_db(Graph)
  381    ).
  382
  383                 /*******************************
  384                 *             LOAD             *
  385                 *******************************/
  386
  387%!  load_db is det.
  388%
  389%   Reload database from the directory specified by rdf_directory/1.
  390%   First we find all names graphs using find_dbs/1 and then we load
  391%   them.
  392
  393load_db :-
  394    rdf_directory(Dir),
  395    concurrency(Jobs),
  396    cpu_stat_key(Jobs, StatKey),
  397    get_time(Wall0),
  398    statistics(StatKey, T0),
  399    load_prefixes(Dir),
  400    verbosity(Silent),
  401    find_dbs(Dir, Graphs, SnapShots, Journals),
  402    length(Graphs, GraphCount),
  403    maplist(rdf_unload_graph, Graphs),
  404    rdf_statistics(triples(Triples0)),
  405    load_sources(snapshots, SnapShots, Silent, Jobs),
  406    load_sources(journals, Journals, Silent, Jobs),
  407    rdf_statistics(triples(Triples1)),
  408    statistics(StatKey, T1),
  409    get_time(Wall1),
  410    T is T1 - T0,
  411    Wall is Wall1 - Wall0,
  412    Triples = Triples1 - Triples0,
  413    message_level(Silent, Level),
  414    print_message(Level, rdf(restore(attached(GraphCount, Triples, T/Wall)))).
  415
  416load_sources(_, [], _, _) :- !.
  417load_sources(Type, Sources, Silent, Jobs) :-
  418    length(Sources, Count),
  419    RunJobs is min(Count, Jobs),
  420    print_message(informational, rdf(restoring(Type, Count, RunJobs))),
  421    make_goals(Sources, Silent, 1, Count, Goals),
  422    concurrent(RunJobs, Goals, []).
  423
  424
  425%!  make_goals(+DBs, +Silent, +Index, +Total, -Goals)
  426
  427make_goals([], _, _, _, []).
  428make_goals([DB|T0], Silent, I,  Total,
  429           [load_source(DB, Silent, I, Total)|T]) :-
  430    I2 is I + 1,
  431    make_goals(T0, Silent, I2, Total, T).
  432
  433verbosity(Silent) :-
  434    rdf_option(silent(Silent)),
  435    !.
  436verbosity(Silent) :-
  437    current_prolog_flag(verbose, silent),
  438    !,
  439    Silent = true.
  440verbosity(brief).
  441
  442
  443%!  concurrency(-Jobs)
  444%
  445%   Number of jobs to run concurrently.
  446
  447concurrency(Jobs) :-
  448    rdf_option(concurrency(Jobs)),
  449    !.
  450concurrency(Jobs) :-
  451    current_prolog_flag(cpu_count, Jobs),
  452    Jobs > 0,
  453    !.
  454concurrency(1).
  455
  456cpu_stat_key(1, cputime) :- !.
  457cpu_stat_key(_, process_cputime).
  458
  459
  460%!  find_dbs(+Dir, -Graphs, -SnapBySize, -JournalBySize) is det.
  461%
  462%   Scan the persistent database and return a list of snapshots and
  463%   journals, both sorted by file-size.  Each term is of the form
  464%
  465%     ==
  466%     db(Size, Ext, DB, DBFile, Depth)
  467%     ==
  468
  469find_dbs(Dir, Graphs, SnapBySize, JournalBySize) :-
  470    directory_files(Dir, Files),
  471    phrase(scan_db_files(Files, Dir, '.', 0), Scanned),
  472    maplist(db_graph, Scanned, UnsortedGraphs),
  473    sort(UnsortedGraphs, Graphs),
  474    (   consider_reindex_db(Dir, Graphs, Scanned)
  475    ->  find_dbs(Dir, Graphs, SnapBySize, JournalBySize)
  476    ;   partition(db_is_snapshot, Scanned, Snapshots, Journals),
  477        sort(Snapshots, SnapBySize),
  478        sort(Journals, JournalBySize)
  479    ).
  480
  481consider_reindex_db(Dir, Graphs, Scanned) :-
  482    length(Graphs, Count),
  483    Count > 0,
  484    DepthNeeded is floor(log(Count)/log(256)),
  485    (   maplist(depth_db(DepthNow), Scanned)
  486    ->  (   DepthNeeded > DepthNow
  487        ->  true
  488        ;   retractall(rdf_option(directory_levels(_))),
  489            assertz(rdf_option(directory_levels(DepthNow))),
  490            fail
  491        )
  492    ;   true
  493    ),
  494    reindex_db(Dir, DepthNeeded).
  495
  496db_is_snapshot(Term) :-
  497    arg(2, Term, trp).
  498
  499db_graph(Term, DB) :-
  500    arg(3, Term, DB).
  501
  502db_file_name(Term, File) :-
  503    arg(4, Term, File).
  504
  505depth_db(Depth, DB) :-
  506    arg(5, DB, Depth).
  507
  508%!  scan_db_files(+Files, +Dir, +Prefix, +Depth)// is det.
  509%
  510%   Produces a list of db(DB,  Size,   File)  for all recognised RDF
  511%   database files.  File is relative to the database directory Dir.
  512
  513scan_db_files([], _, _, _) -->
  514    [].
  515scan_db_files([Nofollow|T], Dir, Prefix, Depth) -->
  516    { nofollow(Nofollow) },
  517    !,
  518    scan_db_files(T, Dir, Prefix, Depth).
  519scan_db_files([File|T], Dir, Prefix, Depth) -->
  520    { file_name_extension(Base, Ext, File),
  521      db_extension(Ext),
  522      !,
  523      rdf_db_to_file(DB, Base),
  524      directory_file_path(Prefix, File, DBFile),
  525      directory_file_path(Dir, DBFile, AbsFile),
  526      size_file(AbsFile, Size)
  527    },
  528    [ db(Size, Ext, DB, AbsFile, Depth) ],
  529    scan_db_files(T, Dir, Prefix, Depth).
  530scan_db_files([D|T], Dir, Prefix, Depth) -->
  531    { directory_file_path(Prefix, D, SubD),
  532      directory_file_path(Dir, SubD, AbsD),
  533      exists_directory(AbsD),
  534      \+ read_link(AbsD, _, _),    % Do not follow links
  535      !,
  536      directory_files(AbsD, SubFiles),
  537      SubDepth is Depth + 1
  538    },
  539    scan_db_files(SubFiles, Dir, SubD, SubDepth),
  540    scan_db_files(T, Dir, Prefix, Depth).
  541scan_db_files([_|T], Dir, Prefix, Depth) -->
  542    scan_db_files(T, Dir, Prefix, Depth).
  543
  544nofollow(.).
  545nofollow(..).
  546
  547db_extension(trp).
  548db_extension(jrn).
  549
  550:- public load_source/4.                % called through make_goals/5
  551
  552load_source(DB, Silent, Nth, Total) :-
  553    db_file_name(DB, File),
  554    db_graph(DB, Graph),
  555    message_level(Silent, Level),
  556    graph_triple_count(Graph, Count0),
  557    statistics(cputime, T0),
  558    (   db_is_snapshot(DB)
  559    ->  print_message(Level, rdf(restore(Silent, snapshot(Graph, File)))),
  560        rdf_load_db(File)
  561    ;   print_message(Level, rdf(restore(Silent, journal(Graph, File)))),
  562        load_journal(File, Graph)
  563    ),
  564    statistics(cputime, T1),
  565    T is T1 - T0,
  566    graph_triple_count(Graph, Count1),
  567    Count is Count1 - Count0,
  568    print_message(Level, rdf(restore(Silent,
  569                                     done(Graph, T, Count, Nth, Total)))).
  570
  571
  572graph_triple_count(Graph, Count) :-
  573    rdf_statistics(triples_by_graph(Graph, Count)),
  574    !.
  575graph_triple_count(_, 0).
  576
  577
  578%!  attach_graph(+Graph, +Options) is det.
  579%
  580%   Load triples and reload  journal   from  the  indicated snapshot
  581%   file.
  582
  583attach_graph(Graph, Options) :-
  584    (   option(silent(true), Options)
  585    ->  Level = silent
  586    ;   Level = informational
  587    ),
  588    db_files(Graph, SnapshotFile, JournalFile),
  589    rdf_retractall(_,_,_,Graph),
  590    statistics(cputime, T0),
  591    print_message(Level, rdf(restore(Silent, Graph))),
  592    db_file(SnapshotFile, AbsSnapShot),
  593    (   exists_file(AbsSnapShot)
  594    ->  print_message(Level, rdf(restore(Silent, snapshot(SnapshotFile)))),
  595        rdf_load_db(AbsSnapShot)
  596    ;   true
  597    ),
  598    (   exists_db(JournalFile)
  599    ->  print_message(Level, rdf(restore(Silent, journal(JournalFile)))),
  600        load_journal(JournalFile, Graph)
  601    ;   true
  602    ),
  603    statistics(cputime, T1),
  604    T is T1 - T0,
  605    (   rdf_statistics(triples_by_graph(Graph, Count))
  606    ->  true
  607    ;   Count = 0
  608    ),
  609    print_message(Level, rdf(restore(Silent,
  610                                     done(Graph, T, Count)))).
  611
  612message_level(true, silent) :- !.
  613message_level(_, informational).
  614
  615
  616                 /*******************************
  617                 *         LOAD JOURNAL         *
  618                 *******************************/
  619
  620%!  load_journal(+File:atom, +DB:atom) is det.
  621%
  622%   Process transactions from the RDF journal File, adding the given
  623%   named graph.
  624
  625load_journal(File, DB) :-
  626    rdf_create_graph(DB),
  627    setup_call_cleanup(
  628        open(File, read, In, [encoding(utf8)]),
  629        ( read(In, T0),
  630          process_journal(T0, In, DB)
  631        ),
  632        close(In)).
  633
  634process_journal(end_of_file, _, _) :- !.
  635process_journal(Term, In, DB) :-
  636    (   process_journal_term(Term, DB)
  637    ->  true
  638    ;   throw(error(type_error(journal_term, Term), _))
  639    ),
  640    read(In, T2),
  641    process_journal(T2, In, DB).
  642
  643process_journal_term(assert(S,P,O), DB) :-
  644    rdf_assert(S,P,O,DB).
  645process_journal_term(assert(S,P,O,Line), DB) :-
  646    rdf_assert(S,P,O,DB:Line).
  647process_journal_term(retract(S,P,O), DB) :-
  648    rdf_retractall(S,P,O,DB).
  649process_journal_term(retract(S,P,O,Line), DB) :-
  650    rdf_retractall(S,P,O,DB:Line).
  651process_journal_term(update(S,P,O,Action), DB) :-
  652    (   rdf_update(S,P,O,DB, Action)
  653    ->  true
  654    ;   print_message(warning, rdf(update_failed(S,P,O,Action)))
  655    ).
  656process_journal_term(start(_), _).      % journal open/close
  657process_journal_term(end(_), _).
  658process_journal_term(begin(_), _).      % logged transaction (compatibility)
  659process_journal_term(end, _).
  660process_journal_term(begin(_,_,_,_), _). % logged transaction (current)
  661process_journal_term(end(_,_,_), _).
  662
  663
  664                 /*******************************
  665                 *         CREATE JOURNAL       *
  666                 *******************************/
  667
  668:- dynamic
  669    blocked_db/2,                   % DB, Reason
  670    transaction_message/3,          % Nesting, Time, Message
  671    transaction_db/3.               % Nesting, DB, Id
  672
  673%!  rdf_persistency(+DB, Bool)
  674%
  675%   Specify whether a database is persistent.  Switching to =false=
  676%   kills the persistent state.  Switching to =true= creates it.
  677
  678rdf_persistency(DB, Bool) :-
  679    must_be(atom, DB),
  680    must_be(boolean, Bool),
  681    fail.
  682rdf_persistency(DB, false) :-
  683    !,
  684    (   blocked_db(DB, persistency)
  685    ->  true
  686    ;   assert(blocked_db(DB, persistency)),
  687        delete_db(DB)
  688    ).
  689rdf_persistency(DB, true) :-
  690    (   retract(blocked_db(DB, persistency))
  691    ->  create_db(DB)
  692    ;   true
  693    ).
  694
  695%!  rdf_db:property_of_graph(?Property, +Graph) is nondet.
  696%
  697%   Extend rdf_graph_property/2 with new properties.
  698
  699:- multifile
  700    rdf_db:property_of_graph/2.  701
  702rdf_db:property_of_graph(persistent(State), Graph) :-
  703    (   blocked_db(Graph, persistency)
  704    ->  State = false
  705    ;   State = true
  706    ).
  707
  708
  709%!  start_monitor is det.
  710%!  stop_monitor is det.
  711%
  712%   Start/stop monitoring the RDF database   for  changes and update
  713%   the journal.
  714
  715start_monitor :-
  716    rdf_monitor(monitor,
  717                [ -assert(load)
  718                ]).
  719stop_monitor :-
  720    rdf_monitor(monitor,
  721                [ -all
  722                ]).
  723
  724%!  monitor(+Term) is semidet.
  725%
  726%   Handle an rdf_monitor/2 callback to  deal with persistency. Note
  727%   that the monitor calls that come   from rdf_db.pl that deal with
  728%   database changes are serialized.  They   do  come from different
  729%   threads though.
  730
  731monitor(Msg) :-
  732    debug(monitor, 'Monitor: ~p~n', [Msg]),
  733    fail.
  734monitor(assert(S,P,O,DB:Line)) :-
  735    !,
  736    \+ blocked_db(DB, _),
  737    journal_fd(DB, Fd),
  738    open_transaction(DB, Fd),
  739    format(Fd, '~q.~n', [assert(S,P,O,Line)]),
  740    sync_journal(DB, Fd).
  741monitor(assert(S,P,O,DB)) :-
  742    \+ blocked_db(DB, _),
  743    journal_fd(DB, Fd),
  744    open_transaction(DB, Fd),
  745    format(Fd, '~q.~n', [assert(S,P,O)]),
  746    sync_journal(DB, Fd).
  747monitor(retract(S,P,O,DB:Line)) :-
  748    !,
  749    \+ blocked_db(DB, _),
  750    journal_fd(DB, Fd),
  751    open_transaction(DB, Fd),
  752    format(Fd, '~q.~n', [retract(S,P,O,Line)]),
  753    sync_journal(DB, Fd).
  754monitor(retract(S,P,O,DB)) :-
  755    \+ blocked_db(DB, _),
  756    journal_fd(DB, Fd),
  757    open_transaction(DB, Fd),
  758    format(Fd, '~q.~n', [retract(S,P,O)]),
  759    sync_journal(DB, Fd).
  760monitor(update(S,P,O,DB:Line,Action)) :-
  761    !,
  762    \+ blocked_db(DB, _),
  763    (   Action = graph(NewDB)
  764    ->  monitor(assert(S,P,O,NewDB)),
  765        monitor(retract(S,P,O,DB:Line))
  766    ;   journal_fd(DB, Fd),
  767        format(Fd, '~q.~n', [update(S,P,O,Action)]),
  768        sync_journal(DB, Fd)
  769    ).
  770monitor(update(S,P,O,DB,Action)) :-
  771    \+ blocked_db(DB, _),
  772    (   Action = graph(NewDB)
  773    ->  monitor(assert(S,P,O,NewDB)),
  774        monitor(retract(S,P,O,DB))
  775    ;   journal_fd(DB, Fd),
  776        open_transaction(DB, Fd),
  777        format(Fd, '~q.~n', [update(S,P,O,Action)]),
  778        sync_journal(DB, Fd)
  779    ).
  780monitor(load(BE, _DumpFileURI)) :-
  781    (   BE = end(Graphs)
  782    ->  sync_loaded_graphs(Graphs)
  783    ;   true
  784    ).
  785monitor(create_graph(Graph)) :-
  786    \+ blocked_db(Graph, _),
  787    journal_fd(Graph, Fd),
  788    open_transaction(Graph, Fd),
  789    sync_journal(Graph, Fd).
  790monitor(reset) :-
  791    forall(rdf_graph(Graph), delete_db(Graph)).
  792                                        % TBD: Remove empty directories?
  793
  794monitor(transaction(BE, Id)) :-
  795    monitor_transaction(Id, BE).
  796
  797monitor_transaction(load_journal(DB), begin(_)) :-
  798    !,
  799    assert(blocked_db(DB, journal)).
  800monitor_transaction(load_journal(DB), end(_)) :-
  801    !,
  802    retractall(blocked_db(DB, journal)).
  803
  804monitor_transaction(parse(URI), begin(_)) :-
  805    !,
  806    (   blocked_db(URI, persistency)
  807    ->  true
  808    ;   assert(blocked_db(URI, parse))
  809    ).
  810monitor_transaction(parse(URI), end(_)) :-
  811    !,
  812    (   retract(blocked_db(URI, parse))
  813    ->  create_db(URI)
  814    ;   true
  815    ).
  816monitor_transaction(unload(DB), begin(_)) :-
  817    !,
  818    (   blocked_db(DB, persistency)
  819    ->  true
  820    ;   assert(blocked_db(DB, unload))
  821    ).
  822monitor_transaction(unload(DB), end(_)) :-
  823    !,
  824    (   retract(blocked_db(DB, unload))
  825    ->  delete_db(DB)
  826    ;   true
  827    ).
  828monitor_transaction(log(Msg), begin(N)) :-
  829    !,
  830    check_nested(N),
  831    get_time(Time),
  832    asserta(transaction_message(N, Time, Msg)).
  833monitor_transaction(log(_), end(N)) :-
  834    check_nested(N),
  835    retract(transaction_message(N, _, _)),
  836    !,
  837    findall(DB:Id, retract(transaction_db(N, DB, Id)), DBs),
  838    end_transactions(DBs, N).
  839monitor_transaction(log(Msg, DB), begin(N)) :-
  840    !,
  841    check_nested(N),
  842    get_time(Time),
  843    asserta(transaction_message(N, Time, Msg)),
  844    journal_fd(DB, Fd),
  845    open_transaction(DB, Fd).
  846monitor_transaction(log(Msg, _DB), end(N)) :-
  847    monitor_transaction(log(Msg), end(N)).
  848
  849
  850%!  check_nested(+Level) is semidet.
  851%
  852%   True if we must log this transaction.   This  is always the case
  853%   for toplevel transactions. Nested transactions   are only logged
  854%   if log_nested_transactions(true) is defined.
  855
  856check_nested(0) :- !.
  857check_nested(_) :-
  858    rdf_option(log_nested_transactions(true)).
  859
  860
  861%!  open_transaction(+DB, +Fd) is det.
  862%
  863%   Add a begin(Id, Level, Time,  Message)   term  if  a transaction
  864%   involves DB. Id is an incremental   integer, where each database
  865%   has its own counter. Level is the nesting level, Time is a floating
  866%   point timestamp and Message is the message provided as argument to
  867%   the log message.
  868
  869open_transaction(DB, Fd) :-
  870    transaction_message(N, Time, Msg),
  871    !,
  872    (   transaction_db(N, DB, _)
  873    ->  true
  874    ;   next_transaction_id(DB, Id),
  875        assert(transaction_db(N, DB, Id)),
  876        RoundedTime is round(Time*100)/100,
  877        format(Fd, '~q.~n', [begin(Id, N, RoundedTime, Msg)])
  878    ).
  879open_transaction(_,_).
  880
  881
  882%!  next_transaction_id(+DB, -Id) is det.
  883%
  884%   Id is the number to user for  the next logged transaction on DB.
  885%   Transactions in each  named  graph   are  numbered  in sequence.
  886%   Searching the Id of the last transaction is performed by the 2nd
  887%   clause starting 1Kb from the end   and doubling this offset each
  888%   failure.
  889
  890:- dynamic
  891    current_transaction_id/2.  892
  893next_transaction_id(DB, Id) :-
  894    retract(current_transaction_id(DB, Last)),
  895    !,
  896    Id is Last + 1,
  897    assert(current_transaction_id(DB, Id)).
  898next_transaction_id(DB, Id) :-
  899    db_files(DB, _, Journal),
  900    exists_file(Journal),
  901    !,
  902    size_file(Journal, Size),
  903    open_db(Journal, read, In, []),
  904    call_cleanup(iterative_expand(In, Size, Last), close(In)),
  905    Id is Last + 1,
  906    assert(current_transaction_id(DB, Id)).
  907next_transaction_id(DB, 1) :-
  908    assert(current_transaction_id(DB, 1)).
  909
  910iterative_expand(_, 0, 0) :- !.
  911iterative_expand(In, Size, Last) :-     % Scan growing sections from the end
  912    Max is floor(log(Size)/log(2)),
  913    between(10, Max, Step),
  914    Offset is -(1<<Step),
  915    seek(In, Offset, eof, _),
  916    skip(In, 10),                   % records are line-based
  917    read(In, T0),
  918    last_transaction_id(T0, In, 0, Last),
  919    Last > 0,
  920    !.
  921iterative_expand(In, _, Last) :-        % Scan the whole file
  922    seek(In, 0, bof, _),
  923    read(In, T0),
  924    last_transaction_id(T0, In, 0, Last).
  925
  926last_transaction_id(end_of_file, _, Last, Last) :- !.
  927last_transaction_id(end(Id, _, _), In, _, Last) :-
  928    read(In, T1),
  929    last_transaction_id(T1, In, Id, Last).
  930last_transaction_id(_, In, Id, Last) :-
  931    read(In, T1),
  932    last_transaction_id(T1, In, Id, Last).
  933
  934
  935%!  end_transactions(+DBs:list(atom:id)) is det.
  936%
  937%   End a transaction that affected the  given list of databases. We
  938%   write the list of other affected databases as an argument to the
  939%   end-term to facilitate fast finding of the related transactions.
  940%
  941%   In each database, the transaction is   ended with a term end(Id,
  942%   Nesting, Others), where  Id  and   Nesting  are  the transaction
  943%   identifier and nesting (see open_transaction/2)  and Others is a
  944%   list of DB:Id,  indicating  other   databases  affected  by  the
  945%   transaction.
  946
  947end_transactions(DBs, N) :-
  948    end_transactions(DBs, DBs, N).
  949
  950end_transactions([], _, _).
  951end_transactions([DB:Id|T], DBs, N) :-
  952    journal_fd(DB, Fd),
  953    once(select(DB:Id, DBs, Others)),
  954    format(Fd, 'end(~q, ~q, ~q).~n', [Id, N, Others]),
  955    sync_journal(DB, Fd),
  956    end_transactions(T, DBs, N).
  957
  958
  959%!  sync_loaded_graphs(+Graphs)
  960%
  961%   Called after a binary triple has been loaded that added triples
  962%   to the given graphs.
  963
  964sync_loaded_graphs(Graphs) :-
  965    maplist(create_db, Graphs).
  966
  967
  968                 /*******************************
  969                 *         JOURNAL FILES        *
  970                 *******************************/
  971
  972%!  journal_fd(+DB, -Stream) is det.
  973%
  974%   Get an open stream to a journal. If the journal is not open, old
  975%   journals are closed to satisfy   the =max_open_journals= option.
  976%   Then the journal is opened in   =append= mode. Journal files are
  977%   always encoded as UTF-8 for  portability   as  well as to ensure
  978%   full coverage of Unicode.
  979
  980journal_fd(DB, Fd) :-
  981    source_journal_fd(DB, Fd),
  982    !.
  983journal_fd(DB, Fd) :-
  984    with_mutex(rdf_journal_file,
  985               journal_fd_(DB, Out)),
  986    Fd = Out.
  987
  988journal_fd_(DB, Fd) :-
  989    source_journal_fd(DB, Fd),
  990    !.
  991journal_fd_(DB, Fd) :-
  992    limit_fd_pool,
  993    db_files(DB, _Snapshot, Journal),
  994    open_db(Journal, append, Fd,
  995            [ close_on_abort(false)
  996            ]),
  997    time_stamp(Now),
  998    format(Fd, '~q.~n', [start([time(Now)])]),
  999    assert(source_journal_fd(DB, Fd)).              % new one at the end
 1000
 1001%!  limit_fd_pool is det.
 1002%
 1003%   Limit the number of  open   journals  to max_open_journals (10).
 1004%   Note that calls  from  rdf_monitor/2   are  issued  in different
 1005%   threads, but as they are part of write operations they are fully
 1006%   synchronised.
 1007
 1008limit_fd_pool :-
 1009    predicate_property(source_journal_fd(_, _), number_of_clauses(N)),
 1010    !,
 1011    (   rdf_option(max_open_journals(Max))
 1012    ->  true
 1013    ;   Max = 10
 1014    ),
 1015    Close is N - Max,
 1016    forall(between(1, Close, _),
 1017           close_oldest_journal).
 1018limit_fd_pool.
 1019
 1020close_oldest_journal :-
 1021    source_journal_fd(DB, _Fd),
 1022    !,
 1023    debug(rdf_persistency, 'Closing old journal for ~q', [DB]),
 1024    close_journal(DB).
 1025close_oldest_journal.
 1026
 1027
 1028%!  sync_journal(+DB, +Fd)
 1029%
 1030%   Sync journal represented by database and   stream.  If the DB is
 1031%   involved in a transaction there is   no point flushing until the
 1032%   end of the transaction.
 1033
 1034sync_journal(DB, _) :-
 1035    transaction_db(_, DB, _),
 1036    !.
 1037sync_journal(_, Fd) :-
 1038    flush_output(Fd).
 1039
 1040%!  close_journal(+DB) is det.
 1041%
 1042%   Close the journal associated with DB if it is open.
 1043
 1044close_journal(DB) :-
 1045    with_mutex(rdf_journal_file,
 1046               close_journal_(DB)).
 1047
 1048close_journal_(DB) :-
 1049    (   retract(source_journal_fd(DB, Fd))
 1050    ->  time_stamp(Now),
 1051        format(Fd, '~q.~n', [end([time(Now)])]),
 1052        close(Fd, [force(true)])
 1053    ;   true
 1054    ).
 1055
 1056%!  close_journals
 1057%
 1058%   Close all open journals.
 1059
 1060close_journals :-
 1061    forall(source_journal_fd(DB, _),
 1062           catch(close_journal(DB), E,
 1063                 print_message(error, E))).
 1064
 1065%!  create_db(+Graph)
 1066%
 1067%   Create a saved version of Graph in corresponding file, close and
 1068%   delete journals.
 1069
 1070create_db(Graph) :-
 1071    \+ rdf(_,_,_,Graph),
 1072    !,
 1073    debug(rdf_persistency, 'Deleting empty Graph ~w', [Graph]),
 1074    delete_db(Graph).
 1075create_db(Graph) :-
 1076    debug(rdf_persistency, 'Saving Graph ~w', [Graph]),
 1077    close_journal(Graph),
 1078    db_abs_files(Graph, Snapshot, Journal),
 1079    atom_concat(Snapshot, '.new', NewSnapshot),
 1080    (   catch(( create_directory_levels(Snapshot),
 1081                rdf_save_db(NewSnapshot, Graph)
 1082              ), Error,
 1083              ( print_message(warning, Error),
 1084                fail
 1085              ))
 1086    ->  (   exists_file(Journal)
 1087        ->  delete_file(Journal)
 1088        ;   true
 1089        ),
 1090        rename_file(NewSnapshot, Snapshot),
 1091        debug(rdf_persistency, 'Saved Graph ~w', [Graph])
 1092    ;   catch(delete_file(NewSnapshot), _, true)
 1093    ).
 1094
 1095
 1096%!  delete_db(+DB)
 1097%
 1098%   Remove snapshot and journal file for DB.
 1099
 1100delete_db(DB) :-
 1101    with_mutex(rdf_journal_file,
 1102               delete_db_(DB)).
 1103
 1104delete_db_(DB) :-
 1105    close_journal_(DB),
 1106    db_abs_files(DB, Snapshot, Journal),
 1107    !,
 1108    (   exists_file(Journal)
 1109    ->  delete_file(Journal)
 1110    ;   true
 1111    ),
 1112    (   exists_file(Snapshot)
 1113    ->  delete_file(Snapshot)
 1114    ;   true
 1115    ).
 1116delete_db_(_).
 1117
 1118                 /*******************************
 1119                 *             LOCKING          *
 1120                 *******************************/
 1121
 1122%!  lock_db(+Dir)
 1123%
 1124%   Lock the database directory Dir.
 1125
 1126lock_db(Dir) :-
 1127    lockfile(Dir, File),
 1128    catch(open(File, update, Out, [lock(write), wait(false)]),
 1129          error(permission_error(Access, _, _), _),
 1130          locked_error(Access, Dir)),
 1131    (   current_prolog_flag(pid, PID)
 1132    ->  true
 1133    ;   PID = 0                     % TBD: Fix in Prolog
 1134    ),
 1135    time_stamp(Now),
 1136    gethostname(Host),
 1137    format(Out, '/* RDF Database is in use */~n~n', []),
 1138    format(Out, '~q.~n', [ locked([ time(Now),
 1139                                    pid(PID),
 1140                                    host(Host)
 1141                                  ])
 1142                         ]),
 1143    flush_output(Out),
 1144    set_end_of_stream(Out),
 1145    assert(rdf_lock(Dir, lock(Out, File))),
 1146    at_halt(unlock_db(Dir)).
 1147
 1148locked_error(lock, Dir) :-
 1149    lockfile(Dir, File),
 1150    (   catch(read_file_to_terms(File, Terms, []), _, fail),
 1151        Terms = [locked(Args)]
 1152    ->  Context = rdf_locked(Args)
 1153    ;   Context = context(_, 'Database is in use')
 1154    ),
 1155    throw(error(permission_error(lock, rdf_db, Dir), Context)).
 1156locked_error(open, Dir) :-
 1157    throw(error(permission_error(lock, rdf_db, Dir),
 1158                context(_, 'Lock file cannot be opened'))).
 1159
 1160%!  unlock_db(+Dir) is det.
 1161%!  unlock_db(+Stream, +File) is det.
 1162
 1163unlock_db(Dir) :-
 1164    retract(rdf_lock(Dir, lock(Out, File))),
 1165    !,
 1166    unlock_db(Out, File).
 1167unlock_db(_).
 1168
 1169unlock_db(Out, File) :-
 1170    close(Out),
 1171    delete_file(File).
 1172
 1173                 /*******************************
 1174                 *           FILENAMES          *
 1175                 *******************************/
 1176
 1177lockfile(Dir, LockFile) :-
 1178    atomic_list_concat([Dir, /, lock], LockFile).
 1179
 1180directory_levels(Levels) :-
 1181    rdf_option(directory_levels(Levels)),
 1182    !.
 1183directory_levels(2).
 1184
 1185db_file(Base, File) :-
 1186    rdf_directory(Dir),
 1187    directory_levels(Levels),
 1188    db_file(Dir, Base, Levels, File).
 1189
 1190db_file(Dir, Base, Levels, File) :-
 1191    dir_levels(Base, Levels, Segments, [Base]),
 1192    atomic_list_concat([Dir|Segments], /, File).
 1193
 1194open_db(Base, Mode, Stream, Options) :-
 1195    db_file(Base, File),
 1196    create_directory_levels(File),
 1197    open(File, Mode, Stream, [encoding(utf8)|Options]).
 1198
 1199create_directory_levels(_File) :-
 1200    rdf_option(directory_levels(0)),
 1201    !.
 1202create_directory_levels(File) :-
 1203    file_directory_name(File, Dir),
 1204    make_directory_path(Dir).
 1205
 1206exists_db(Base) :-
 1207    db_file(Base, File),
 1208    exists_file(File).
 1209
 1210%!  dir_levels(+File, +Levels, ?Segments, ?Tail) is det.
 1211%
 1212%   Create a list of intermediate directory names for File.  Each
 1213%   directory consists of two hexadecimal digits.
 1214
 1215dir_levels(_, 0, Segments, Segments) :- !.
 1216dir_levels(File, Levels, Segments, Tail) :-
 1217    rdf_atom_md5(File, 1, Hash),
 1218    create_dir_levels(Levels, 0, Hash, Segments, Tail).
 1219
 1220create_dir_levels(0, _, _, Segments, Segments) :- !.
 1221create_dir_levels(N, S, Hash, [S1|Segments0], Tail) :-
 1222    sub_atom(Hash, S, 2, _, S1),
 1223    S2 is S+2,
 1224    N2 is N-1,
 1225    create_dir_levels(N2, S2, Hash, Segments0, Tail).
 1226
 1227
 1228%!  db_files(+DB, -Snapshot, -Journal).
 1229%!  db_files(-DB, +Snapshot, -Journal).
 1230%!  db_files(-DB, -Snapshot, +Journal).
 1231%
 1232%   True if named graph DB is represented  by the files Snapshot and
 1233%   Journal. The filenames are local   to the directory representing
 1234%   the store.
 1235
 1236db_files(DB, Snapshot, Journal) :-
 1237    nonvar(DB),
 1238    !,
 1239    rdf_db_to_file(DB, Base),
 1240    atom_concat(Base, '.trp', Snapshot),
 1241    atom_concat(Base, '.jrn', Journal).
 1242db_files(DB, Snapshot, Journal) :-
 1243    nonvar(Snapshot),
 1244    !,
 1245    atom_concat(Base, '.trp', Snapshot),
 1246    atom_concat(Base, '.jrn', Journal),
 1247    rdf_db_to_file(DB, Base).
 1248db_files(DB, Snapshot, Journal) :-
 1249    nonvar(Journal),
 1250    !,
 1251    atom_concat(Base, '.jrn', Journal),
 1252    atom_concat(Base, '.trp', Snapshot),
 1253    rdf_db_to_file(DB, Base).
 1254
 1255db_abs_files(DB, Snapshot, Journal) :-
 1256    db_files(DB, Snapshot0, Journal0),
 1257    db_file(Snapshot0, Snapshot),
 1258    db_file(Journal0, Journal).
 1259
 1260
 1261%!  rdf_journal_file(+Graph, -File) is semidet.
 1262%!  rdf_journal_file(-Graph, -File) is nondet.
 1263%
 1264%   True if File the name of the existing journal file for Graph.
 1265
 1266rdf_journal_file(Graph, Journal) :-
 1267    (   var(Graph)
 1268    ->  rdf_graph(Graph)
 1269    ;   true
 1270    ),
 1271    db_abs_files(Graph, _Snapshot, Journal),
 1272    exists_file(Journal).
 1273
 1274
 1275%!  rdf_snapshot_file(+Graph, -File) is semidet.
 1276%!  rdf_snapshot_file(-Graph, -File) is nondet.
 1277%
 1278%   True if File the name of the existing snapshot file for Graph.
 1279
 1280rdf_snapshot_file(Graph, Snapshot) :-
 1281    (   var(Graph)
 1282    ->  rdf_graph(Graph)    % also pick the empty graphs
 1283    ;   true
 1284    ),
 1285    db_abs_files(Graph, Snapshot, _Journal),
 1286    exists_file(Snapshot).
 1287
 1288
 1289%!  rdf_db_to_file(+DB, -File) is det.
 1290%!  rdf_db_to_file(-DB, +File) is det.
 1291%
 1292%   Translate between database encoding (often an   file or URL) and
 1293%   the name we store in the  directory.   We  keep  a cache for two
 1294%   reasons. Speed, but much more important   is that the mapping of
 1295%   raw --> encoded provided by  www_form_encode/2 is not guaranteed
 1296%   to be unique by the W3C standards.
 1297
 1298rdf_db_to_file(DB, File) :-
 1299    file_base_db(File, DB),
 1300    !.
 1301rdf_db_to_file(DB, File) :-
 1302    url_to_filename(DB, File),
 1303    assert(file_base_db(File, DB)).
 1304
 1305%!  url_to_filename(+URL, -FileName) is det.
 1306%!  url_to_filename(-URL, +FileName) is det.
 1307%
 1308%   Turn  a  valid  URL  into  a  filename.  Earlier  versions  used
 1309%   www_form_encode/2, but this can produce  characters that are not
 1310%   valid  in  filenames.  We  will  use    the   same  encoding  as
 1311%   www_form_encode/2,  but  using  our  own    rules   for  allowed
 1312%   characters. The only requirement is that   we avoid any filename
 1313%   special character in use.  The   current  encoding  use US-ASCII
 1314%   alnum characters, _ and %
 1315
 1316url_to_filename(URL, FileName) :-
 1317    atomic(URL),
 1318    !,
 1319    atom_codes(URL, Codes),
 1320    phrase(url_encode(EncCodes), Codes),
 1321    atom_codes(FileName, EncCodes).
 1322url_to_filename(URL, FileName) :-
 1323    uri_encoded(path, URL, FileName).
 1324
 1325url_encode([0'+|T]) -->
 1326    " ",
 1327    !,
 1328    url_encode(T).
 1329url_encode([C|T]) -->
 1330    alphanum(C),
 1331    !,
 1332    url_encode(T).
 1333url_encode([C|T]) -->
 1334    no_enc_extra(C),
 1335    !,
 1336    url_encode(T).
 1337url_encode(Enc) -->
 1338    (   "\r\n"
 1339    ;   "\n"
 1340    ),
 1341    !,
 1342    { string_codes("%0D%0A", Codes),
 1343      append(Codes, T, Enc)
 1344    },
 1345    url_encode(T).
 1346url_encode([]) -->
 1347    eos,
 1348    !.
 1349url_encode([0'%,D1,D2|T]) -->
 1350    [C],
 1351    { Dv1 is (C>>4 /\ 0xf),
 1352      Dv2 is (C /\ 0xf),
 1353      code_type(D1, xdigit(Dv1)),
 1354      code_type(D2, xdigit(Dv2))
 1355    },
 1356    url_encode(T).
 1357
 1358eos([], []).
 1359
 1360alphanum(C) -->
 1361    [C],
 1362    { C < 128,                      % US-ASCII
 1363      code_type(C, alnum)
 1364    }.
 1365
 1366no_enc_extra(0'_) --> "_".
 1367
 1368
 1369                 /*******************************
 1370                 *             REINDEX          *
 1371                 *******************************/
 1372
 1373%!  reindex_db(+Dir, +Levels)
 1374%
 1375%   Reindex the database by creating intermediate directories.
 1376
 1377reindex_db(Dir, Levels) :-
 1378    directory_files(Dir, Files),
 1379    reindex_files(Files, Dir, '.', 0, Levels),
 1380    remove_empty_directories(Files, Dir).
 1381
 1382reindex_files([], _, _, _, _).
 1383reindex_files([Nofollow|Files], Dir, Prefix, CLevel, Levels) :-
 1384    nofollow(Nofollow),
 1385    !,
 1386    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1387reindex_files([File|Files], Dir, Prefix, CLevel, Levels) :-
 1388    CLevel \== Levels,
 1389    file_name_extension(_Base, Ext, File),
 1390    db_extension(Ext),
 1391    !,
 1392    directory_file_path(Prefix, File, DBFile),
 1393    directory_file_path(Dir, DBFile, OldPath),
 1394    db_file(Dir, File, Levels, NewPath),
 1395    debug(rdf_persistency, 'Rename ~q --> ~q', [OldPath, NewPath]),
 1396    file_directory_name(NewPath, NewDir),
 1397    make_directory_path(NewDir),
 1398    rename_file(OldPath, NewPath),
 1399    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1400reindex_files([D|Files], Dir, Prefix, CLevel, Levels) :-
 1401    directory_file_path(Prefix, D, SubD),
 1402    directory_file_path(Dir, SubD, AbsD),
 1403    exists_directory(AbsD),
 1404    \+ read_link(AbsD, _, _),      % Do not follow links
 1405    !,
 1406    directory_files(AbsD, SubFiles),
 1407    CLevel2 is CLevel + 1,
 1408    reindex_files(SubFiles, Dir, SubD, CLevel2, Levels),
 1409    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1410reindex_files([_|Files], Dir, Prefix, CLevel, Levels) :-
 1411    reindex_files(Files, Dir, Prefix, CLevel, Levels).
 1412
 1413
 1414remove_empty_directories([], _).
 1415remove_empty_directories([File|Files], Dir) :-
 1416    \+ nofollow(File),
 1417    directory_file_path(Dir, File, Path),
 1418    exists_directory(Path),
 1419    \+ read_link(Path, _, _),
 1420    !,
 1421    directory_files(Path, Content),
 1422    exclude(nofollow, Content, RealContent),
 1423    (   RealContent == []
 1424    ->  debug(rdf_persistency, 'Remove empty dir ~q', [Path]),
 1425        delete_directory(Path)
 1426    ;   remove_empty_directories(RealContent, Path)
 1427    ),
 1428    remove_empty_directories(Files, Dir).
 1429remove_empty_directories([_|Files], Dir) :-
 1430    remove_empty_directories(Files, Dir).
 1431
 1432
 1433                 /*******************************
 1434                 *            PREFIXES          *
 1435                 *******************************/
 1436
 1437save_prefixes(Dir) :-
 1438    atomic_list_concat([Dir, /, 'prefixes.db'], PrefixFile),
 1439    setup_call_cleanup(open(PrefixFile, write, Out, [encoding(utf8)]),
 1440                       write_prefixes(Out),
 1441                       close(Out)).
 1442
 1443write_prefixes(Out) :-
 1444    format(Out, '% Snapshot of defined RDF prefixes~n~n', []),
 1445    forall(rdf_current_ns(Alias, URI),
 1446           format(Out, 'prefix(~q, ~q).~n', [Alias, URI])).
 1447
 1448%!  load_prefixes(+RDFDBDir) is det.
 1449%
 1450%   If the file RDFDBDir/prefixes.db exists,  load the prefixes. The
 1451%   prefixes are registered using rdf_register_ns/3. Possible errors
 1452%   because the prefix  definitions  have   changed  are  printed as
 1453%   warnings, retaining the  old  definition.   Note  that  changing
 1454%   prefixes generally requires reloading all RDF from the source.
 1455
 1456load_prefixes(Dir) :-
 1457    atomic_list_concat([Dir, /, 'prefixes.db'], PrefixFile),
 1458    (   exists_file(PrefixFile)
 1459    ->  setup_call_cleanup(open(PrefixFile, read, In, [encoding(utf8)]),
 1460                           read_prefixes(In),
 1461                           close(In))
 1462    ;   true
 1463    ).
 1464
 1465read_prefixes(Stream) :-
 1466    read_term(Stream, T0, []),
 1467    read_prefixes(T0, Stream).
 1468
 1469read_prefixes(end_of_file, _) :- !.
 1470read_prefixes(prefix(Alias, URI), Stream) :-
 1471    !,
 1472    must_be(atom, Alias),
 1473    must_be(atom, URI),
 1474    catch(rdf_register_ns(Alias, URI, []), E,
 1475          print_message(warning, E)),
 1476    read_term(Stream, T, []),
 1477    read_prefixes(T, Stream).
 1478read_prefixes(Term, _) :-
 1479    domain_error(prefix_term, Term).
 1480
 1481
 1482                 /*******************************
 1483                 *              UTIL            *
 1484                 *******************************/
 1485
 1486%!  mkdir(+Directory)
 1487%
 1488%   Create a directory if it does not already exist.
 1489
 1490mkdir(Directory) :-
 1491    exists_directory(Directory),
 1492    !.
 1493mkdir(Directory) :-
 1494    make_directory(Directory).
 1495
 1496%!  time_stamp(-Integer)
 1497%
 1498%   Return time-stamp rounded to integer.
 1499
 1500time_stamp(Int) :-
 1501    get_time(Now),
 1502    Int is round(Now).
 1503
 1504
 1505                 /*******************************
 1506                 *            MESSAGES          *
 1507                 *******************************/
 1508
 1509:- multifile
 1510    prolog:message/3,
 1511    prolog:message_context/3. 1512
 1513prolog:message(rdf(Term)) -->
 1514    message(Term).
 1515
 1516message(restoring(Type, Count, Jobs)) -->
 1517    [ 'Restoring ~D ~w using ~D concurrent workers'-[Count, Type, Jobs] ].
 1518message(restore(attached(Graphs, Triples, Time/Wall))) -->
 1519    { catch(Percent is round(100*Time/Wall), _, Percent = 0) },
 1520    [ 'Loaded ~D graphs (~D triples) in ~2f sec. (~d% CPU = ~2f sec.)'-
 1521      [Graphs, Triples, Wall, Percent, Time] ].
 1522% attach_graph/2
 1523message(restore(true, Action)) -->
 1524    !,
 1525    silent_message(Action).
 1526message(restore(brief, Action)) -->
 1527    !,
 1528    brief_message(Action).
 1529message(restore(_, Graph)) -->
 1530    [ 'Restoring ~p ... '-[Graph], flush ].
 1531message(restore(_, snapshot(_))) -->
 1532    [ at_same_line, '(snapshot) '-[], flush ].
 1533message(restore(_, journal(_))) -->
 1534    [ at_same_line, '(journal) '-[], flush ].
 1535message(restore(_, done(_, Time, Count))) -->
 1536    [ at_same_line, '~D triples in ~2f sec.'-[Count, Time] ].
 1537% load_source/4
 1538message(restore(_, snapshot(G, _))) -->
 1539    [ 'Restoring ~p\t(snapshot)'-[G], flush ].
 1540message(restore(_, journal(G, _))) -->
 1541    [ 'Restoring ~p\t(journal)'-[G], flush ].
 1542message(restore(_, done(_, Time, Count))) -->
 1543    [ at_same_line, '~D triples in ~2f sec.'-[Count, Time] ].
 1544% journal handling
 1545message(update_failed(S,P,O,Action)) -->
 1546    [ 'Failed to update <~p ~p ~p> with ~p'-[S,P,O,Action] ].
 1547% directory reindexing
 1548message(reindex(Count, Depth)) -->
 1549    [ 'Restructuring database with ~d levels (~D graphs)'-[Depth, Count] ].
 1550message(reindex(Depth)) -->
 1551    [ 'Fixing database directory structure (~d levels)'-[Depth] ].
 1552message(read_only) -->
 1553    [ 'Cannot write persistent store; continuing in read-only mode.', nl,
 1554      'All changes to the RDF store will be lost if this process terminates.'
 1555    ].
 1556
 1557silent_message(_Action) --> [].
 1558
 1559brief_message(done(Graph, _Time, _Count, Nth, Total)) -->
 1560    { file_base_name(Graph, Base) },
 1561    [ at_same_line,
 1562      '\r~p~`.t ~D of ~D graphs~72|'-[Base, Nth, Total],
 1563      flush
 1564    ].
 1565brief_message(_) --> [].
 1566
 1567
 1568prolog:message_context(rdf_locked(Args)) -->
 1569    { memberchk(time(Time), Args),
 1570      memberchk(pid(Pid), Args),
 1571      format_time(string(S), '%+', Time)
 1572    },
 1573    [ nl,
 1574      'locked at ~s by process id ~w'-[S,Pid]
 1575    ]